{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Agenda\n",
    "\n",
    "- 2005 Surdirman Cup\n",
    "- Code examples\n",
    "  - Hub\n",
    "  - Chat\n",
    "- Traps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "例题：苏迪曼杯积分系统开发\n",
    "- 项目背景：2005年苏迪曼杯（羽毛球混合团体赛）积分系统开发项目，由陈硕研究生时期参与开发\n",
    "- 技术特点：使用TCP协议向多个客户端进行广播的聊天服务器实现\n",
    "- 示例类型：包含Hub（基于主题的广播）和Chat（聊天服务器）两个代码示例\n",
    "- 开发陷阱：TCP用户态广播存在多个易错点需要特别注意"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "项目背景与开发人员\n",
    "\n",
    "- 开发团队：陈硕与实验室刘老师共同开发\n",
    "- 项目验证：2005年4月29日在国家体育总局训练局进行系统试运行\n",
    "- 接触人员：包括林丹（男子单打冠军）、张军（奥运会双打冠军）等羽毛球名将"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<img src=\"./images/8.1.png\" alt=\"8.1\" style=\"zoom:43%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "场地与显示设备配置\n",
    "\n",
    "- 场地布局：预赛阶段7片场地，半决赛/决赛缩减为1片主场地\n",
    "- 显示配置：\n",
    "  - 预赛：每片场地对角布置2块42寸等离子显示器（共14块）\n",
    "  - 决赛：主场地四角各1块显示器（共4块）\n",
    "- 赛程特点：对阵队伍提前确定，具体出场队员赛前几小时确定"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/8.2.png\" alt=\"8.2\" style=\"zoom:43%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "决赛与半决赛的显示设备\n",
    "\n",
    "- 界面内容：显示三局比分、代表队名称、出场运动员姓名（如张军/高凌组合）\n",
    "- 设备配置：笔记本外接等离子显示器，现场照片显示开发人员位于场地右下角"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "开发设备与环境\n",
    "\n",
    "- 硬件方案：使用普通PC开发，笔记本和显示器均为租赁设备\n",
    "- 备用方案：决赛时准备双机热备（一台运行一台备用）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/8.3.png\" alt=\"8.3\" style=\"zoom:43%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "比赛记录与大屏幕显示\n",
    "\n",
    "- 辅助显示：赛场两端设置大屏幕轮流显示各场地比分\n",
    "- 显示技术：采用红绿双色显示，需配合蓝色背景避免颜色偏差\n",
    "- 视频输出：使用模拟视频信号输出（类似老式电视技术）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Taufik Hidayat vs Lin Dan"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 历史比赛：2005年苏迪曼杯决赛林丹对阵陶菲克·希达亚特\n",
    "- 系统验证：开发人员现场保障系统运行，同时观看比赛"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Requirements"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 7 courts, each with 2 screens\n",
    "  - Final: one court with four screens\n",
    "- One big screen shows all 7 courts in turn\n",
    "- One database\n",
    "- Upload to Web via FTP\n",
    "- No single point of failure (SPoF)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心需求：\n",
    "  - 支持7片场地×2屏幕→决赛1片场地×4屏幕的灵活配置\n",
    "  - 大屏幕轮播所有场地比分\n",
    "  - 中央数据库存储赛程和出场信息\n",
    "  - FTP实时上传比分至网易体育频道\n",
    "- 可靠性：严格避免单点故障(SPoF)，确保比赛不受设备故障影响\n",
    "- 特殊需求：淘汰制比赛需要动态更新晋级信息"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Broadcasting"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 重复 06_Simple memcached"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "羽毛球竞赛系统设计考虑\n",
    "\n",
    "- 弃权处理机制：系统设计时考虑了队员比赛中弃权的情况（如丹麦女队崴脚弃权案例），但未考虑临时换场地需求（如场地不平需更换场地的情况）\n",
    "- 赛制变更影响：原系统基于15分换发球制开发，后羽毛球改为21分制（取消发球权规则），导致系统逻辑完全失效\n",
    "- 硬件配置：采用机械积分牌作为临时解决方案，该问题仅出现在预赛阶段，决赛场地不会发生"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "羽毛球竞赛系统结构\n",
    "\n",
    "- 场地配置：每片场地配备显示器和电脑（共7套），中央设置数据库服务器存储比赛进度\n",
    "- 控制终端：\n",
    "  - 大屏幕控制电脑（控制两端大屏幕）\n",
    "  - 数据录入电脑（记录队员出场顺序）\n",
    "  - FTP上传电脑（向网易体育频道传输比分数据）\n",
    "- 容错设计：\n",
    "  - 比分记录席独立运行，电源中断不影响比赛进行\n",
    "  - 各场地系统相互隔离（如4号场地故障不影响3号场地）\n",
    "- 计分设备演进：\n",
    "  - 原计划使用无线PDA，因现场无线干扰改为有线USB连接\n",
    "  - PDA电池续航不足（全天比赛需频繁充电）\n",
    "  - 计分软件统一使用C#开发"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "积分牌开发系统\n",
    "\n",
    "- 核心机制：采用发布/订阅模式（Pub/Sub）实现比分广播\n",
    "- 数据格式：\n",
    "  - Snapshot：完整当前比分（如中国5:2）\n",
    "  - Delta：增量更新（如中国队+1分）\n",
    "- 系统优势：\n",
    "  - 组件独立性高，支持单点重启\n",
    "  - 采用Snapshot机制避免历史记录依赖\n",
    "  - 异常处理：PDA和HUB同时故障时从数据库恢复最后比分\n",
    "- 特殊需求实现：为境外转播方提供独立比分接入方案（绕过中文体育频道）\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "分布式系统的不可靠性\n",
    "\n",
    "- 典型案例：泰国队抗议事件（对手违反出场顺序规则），导致比赛结果悬而未决\n",
    "- 系统应对：数据库支持未决状态记录，不影响其他比赛正常进行"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "广播机制的选择与实现\n",
    "\n",
    "- 协议选择：\n",
    "  - 放弃IP广播（仅UDP支持，调试困难）\n",
    "  - 采用TCP实现应用层广播（端到端可调试）\n",
    "- 架构设计：\n",
    "  - 每个场地独立HUB进程\n",
    "  - 单发布者+多订阅者模式（保证比分修改唯一性）\n",
    "  - 主从备份设计（master-slave配对）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "系统性能考量\n",
    "\n",
    "- 实时性要求：羽毛球比分变化间隔约几秒（不同于乒乓球的高速变化）\n",
    "- 开发难点：\n",
    "  - HUB实现（单点接收，多点分发）\n",
    "  - 协议一致性维护\n",
    "  - 故障恢复机制设计"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Chat server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Four versions\n",
    "  - A. Single thread (server.cc)\n",
    "  - B. Multi threaded, one loop per thread, server_threaded.cc\n",
    "  - C. B + reduce critical section , server_threaded_efficient.cc\n",
    "  - D. B + thread local , server_threaded_highperformance.cc\n",
    "- Section 7.3 and 2.8 of my book\n",
    "- Benchmark: https://github.com/tolbrino/hotwheels\n",
    "- https://github.com/chenshuo/muduo/tree/master/examples/asio/chat"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "聊天服务器\n",
    "\n",
    "- 版本演进:\n",
    "  - A. 单线程版本(server.cc)\n",
    "  - B. 多线程版本，每个线程一个事件循环(server_threaded.cc)\n",
    "  - C. B版本+减少临界区(server_threaded_efficient.cc)\n",
    "  - D. B版本+线程本地存储(server_threaded_highperformance.cc): 将消息广播到4个线程, 每个线程只对自己的连接发, 不需要全局锁\n",
    "- 性能基准:\n",
    "  - 测试工具：hotwheels项目(基于libuv)\n",
    "  - 测试场景：1万个TCP并发连接下广播消息的延迟\n",
    "  - 预期性能：D版本最佳，A版本最差"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/loadtest.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Atomic.h\"\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/EventLoopThreadPool.h\"\n",
    "#include \"muduo/net/TcpClient.h\"\n",
    "\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "int g_connections = 0;\n",
    "AtomicInt32 g_aliveConnections;\n",
    "AtomicInt32 g_messagesReceived;\n",
    "Timestamp g_startTime;\n",
    "std::vector<Timestamp> g_receiveTime;\n",
    "EventLoop* g_loop;\n",
    "std::function<void()> g_statistic;\n",
    "\n",
    "class ChatClient : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatClient(EventLoop* loop, const InetAddress& serverAddr)\n",
    "    : loop_(loop),\n",
    "      client_(loop, serverAddr, \"LoadTestClient\"),\n",
    "      codec_(std::bind(&ChatClient::onStringMessage, this, _1, _2, _3))\n",
    "  {\n",
    "    client_.setConnectionCallback(\n",
    "        std::bind(&ChatClient::onConnection, this, _1));\n",
    "    client_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "    //client_.enableRetry();\n",
    "  }\n",
    "\n",
    "  void connect()\n",
    "  {\n",
    "    client_.connect();\n",
    "  }\n",
    "\n",
    "  void disconnect()\n",
    "  {\n",
    "    // client_.disconnect();\n",
    "  }\n",
    "\n",
    "  Timestamp receiveTime() const { return receiveTime_; }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->localAddress().toIpPort() << \" -> \"\n",
    "             << conn->peerAddress().toIpPort() << \" is \"\n",
    "             << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      connection_ = conn;\n",
    "      if (g_aliveConnections.incrementAndGet() == g_connections)\n",
    "      {\n",
    "        LOG_INFO << \"all connected\";\n",
    "        loop_->runAfter(10.0, std::bind(&ChatClient::send, this));\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      connection_.reset();\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    // printf(\"<<< %s\\n\", message.c_str());\n",
    "    receiveTime_ = loop_->pollReturnTime();\n",
    "    int received = g_messagesReceived.incrementAndGet();\n",
    "    if (received == g_connections)\n",
    "    {\n",
    "      Timestamp endTime = Timestamp::now();\n",
    "      LOG_INFO << \"all received \" << g_connections << \" in \"\n",
    "               << timeDifference(endTime, g_startTime);\n",
    "      g_loop->queueInLoop(g_statistic);\n",
    "    }\n",
    "    else if (received % 1000 == 0)\n",
    "    {\n",
    "      LOG_DEBUG << received;\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void send()\n",
    "  {\n",
    "    g_startTime = Timestamp::now();\n",
    "    codec_.send(get_pointer(connection_), \"hello\");\n",
    "    LOG_DEBUG << \"sent\";\n",
    "  }\n",
    "\n",
    "  EventLoop* loop_;\n",
    "  TcpClient client_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  TcpConnectionPtr connection_;\n",
    "  Timestamp receiveTime_;\n",
    "};\n",
    "\n",
    "void statistic(const std::vector<std::unique_ptr<ChatClient>>& clients)\n",
    "{\n",
    "  LOG_INFO << \"statistic \" << clients.size();\n",
    "  std::vector<double> seconds(clients.size());\n",
    "  for (size_t i = 0; i < clients.size(); ++i)\n",
    "  {\n",
    "    seconds[i] = timeDifference(clients[i]->receiveTime(), g_startTime);\n",
    "  }\n",
    "\n",
    "  std::sort(seconds.begin(), seconds.end());\n",
    "  for (size_t i = 0; i < clients.size(); i += std::max(static_cast<size_t>(1), clients.size()/20))\n",
    "  {\n",
    "    printf(\"%6zd%% %.6f\\n\", i*100/clients.size(), seconds[i]);\n",
    "  }\n",
    "  if (clients.size() >= 100)\n",
    "  {\n",
    "    printf(\"%6d%% %.6f\\n\", 99, seconds[clients.size() - clients.size()/100]);\n",
    "  }\n",
    "  printf(\"%6d%% %.6f\\n\", 100, seconds.back());\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 3)\n",
    "  {\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[2]));\n",
    "    InetAddress serverAddr(argv[1], port);\n",
    "    g_connections = atoi(argv[3]);\n",
    "    int threads = 0;\n",
    "    if (argc > 4)\n",
    "    {\n",
    "      threads = atoi(argv[4]);\n",
    "    }\n",
    "\n",
    "    EventLoop loop;\n",
    "    g_loop = &loop;\n",
    "    EventLoopThreadPool loopPool(&loop, \"chat-loadtest\");\n",
    "    loopPool.setThreadNum(threads);\n",
    "    loopPool.start();\n",
    "\n",
    "    g_receiveTime.reserve(g_connections);\n",
    "    std::vector<std::unique_ptr<ChatClient>> clients(g_connections);\n",
    "    g_statistic = std::bind(statistic, std::ref(clients));\n",
    "\n",
    "    for (int i = 0; i < g_connections; ++i)\n",
    "    {\n",
    "      clients[i].reset(new ChatClient(loopPool.getNextLoop(), serverAddr));\n",
    "      clients[i]->connect();\n",
    "      usleep(200);\n",
    "    }\n",
    "\n",
    "    loop.loop();\n",
    "    // client.disconnect();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s host_ip port connections [threads]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "负载测试\n",
    "- 测试目的：对比四种服务器实现的性能差异\n",
    "- 实现方式：\n",
    "  - 使用`boost::ptr_vector`管理测试客户端\n",
    "  - 记录每个客户端接收消息的时间戳\n",
    "  - 计算消息广播的延迟分布"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/codec.h\n",
    "\n",
    "#ifndef MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H\n",
    "#define MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/net/Buffer.h\"\n",
    "#include \"muduo/net/Endian.h\"\n",
    "#include \"muduo/net/TcpConnection.h\"\n",
    "\n",
    "class LengthHeaderCodec : muduo::noncopyable\n",
    "{\n",
    " public:\n",
    "  typedef std::function<void (const muduo::net::TcpConnectionPtr&,\n",
    "                                const muduo::string& message,\n",
    "                                muduo::Timestamp)> StringMessageCallback;\n",
    "\n",
    "  explicit LengthHeaderCodec(const StringMessageCallback& cb)\n",
    "    : messageCallback_(cb)\n",
    "  {\n",
    "  }\n",
    "\n",
    "  void onMessage(const muduo::net::TcpConnectionPtr& conn,\n",
    "                 muduo::net::Buffer* buf,\n",
    "                 muduo::Timestamp receiveTime)\n",
    "  {\n",
    "    while (buf->readableBytes() >= kHeaderLen) // kHeaderLen == 4\n",
    "    {\n",
    "      // FIXME: use Buffer::peekInt32()\n",
    "      const void* data = buf->peek();\n",
    "      int32_t be32 = *static_cast<const int32_t*>(data); // SIGBUS\n",
    "      const int32_t len = muduo::net::sockets::networkToHost32(be32);\n",
    "      if (len > 65536 || len < 0)\n",
    "      {\n",
    "        LOG_ERROR << \"Invalid length \" << len;\n",
    "        conn->shutdown();  // FIXME: disable reading\n",
    "        break;\n",
    "      }\n",
    "      else if (buf->readableBytes() >= len + kHeaderLen)\n",
    "      {\n",
    "        buf->retrieve(kHeaderLen);\n",
    "        muduo::string message(buf->peek(), len);\n",
    "        messageCallback_(conn, message, receiveTime);\n",
    "        buf->retrieve(len);\n",
    "      }\n",
    "      else\n",
    "      {\n",
    "        break;\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "\n",
    "  // FIXME: TcpConnectionPtr\n",
    "  void send(muduo::net::TcpConnection* conn,\n",
    "            const muduo::StringPiece& message)\n",
    "  {\n",
    "    muduo::net::Buffer buf;\n",
    "    buf.append(message.data(), message.size());\n",
    "    int32_t len = static_cast<int32_t>(message.size());\n",
    "    int32_t be32 = muduo::net::sockets::hostToNetwork32(len);\n",
    "    buf.prepend(&be32, sizeof be32);\n",
    "    conn->send(&buf);\n",
    "  }\n",
    "\n",
    " private:\n",
    "  StringMessageCallback messageCallback_;\n",
    "  const static size_t kHeaderLen = sizeof(int32_t);\n",
    "};\n",
    "\n",
    "#endif  // MUDUO_EXAMPLES_ASIO_CHAT_CODEC_H"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 协议格式：\n",
    "  - 固定4字节长度头(网络字节序)\n",
    "  - 最大消息长度限制为65536字节\n",
    "- 解码过程：\n",
    "  - 检查缓冲区可读数据是否足够\n",
    "  - 转换网络字节序到主机字节序\n",
    "  - 验证长度有效性\n",
    "- 编码过程：\n",
    "  - 在消息前添加4字节长度头\n",
    "  - 转换主机字节序到网络字节序"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/server.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/TcpServer.h\"\n",
    "\n",
    "#include <set>\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "class ChatServer : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatServer(EventLoop* loop,\n",
    "             const InetAddress& listenAddr)\n",
    "  : server_(loop, listenAddr, \"ChatServer\"),\n",
    "    codec_(std::bind(&ChatServer::onStringMessage, this, _1, _2, _3)) // message变为string message\n",
    "  {\n",
    "    server_.setConnectionCallback(\n",
    "        std::bind(&ChatServer::onConnection, this, _1));\n",
    "    server_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "  }\n",
    "\n",
    "  void start()\n",
    "  {\n",
    "    server_.start();\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->peerAddress().toIpPort() << \" -> \"\n",
    "             << conn->localAddress().toIpPort() << \" is \"\n",
    "             << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      connections_.insert(conn);\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      connections_.erase(conn);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    for (ConnectionList::iterator it = connections_.begin(); // 业务逻辑\n",
    "        it != connections_.end();\n",
    "        ++it)\n",
    "    {\n",
    "      codec_.send(get_pointer(*it), message);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  typedef std::set<TcpConnectionPtr> ConnectionList;\n",
    "  TcpServer server_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  ConnectionList connections_;\n",
    "};\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 1)\n",
    "  {\n",
    "    EventLoop loop;\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));\n",
    "    InetAddress serverAddr(port);\n",
    "    ChatServer server(&loop, serverAddr);\n",
    "    server.start();\n",
    "    loop.loop();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s port\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心结构：\n",
    "  - 使用`std::set`保存所有TCP连接\n",
    "  - 基于事件循环处理网络IO\n",
    "- 业务逻辑：\n",
    "  - 收到消息后广播给所有连接(包括发送者)\n",
    "  - 仅需10行核心业务代码\n",
    "- 性能特点：\n",
    "  - 无锁设计\n",
    "  - 简单但扩展性有限"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/server_threaded.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/TcpServer.h\"\n",
    "\n",
    "#include <set>\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "class ChatServer : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatServer(EventLoop* loop,\n",
    "             const InetAddress& listenAddr)\n",
    "  : server_(loop, listenAddr, \"ChatServer\"),\n",
    "    codec_(std::bind(&ChatServer::onStringMessage, this, _1, _2, _3))\n",
    "  {\n",
    "    server_.setConnectionCallback(\n",
    "        std::bind(&ChatServer::onConnection, this, _1));\n",
    "    server_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "  }\n",
    "\n",
    "  void setThreadNum(int numThreads)\n",
    "  {\n",
    "    server_.setThreadNum(numThreads);\n",
    "  }\n",
    "\n",
    "  void start()\n",
    "  {\n",
    "    server_.start();\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->peerAddress().toIpPort() << \" -> \"\n",
    "        << conn->localAddress().toIpPort() << \" is \"\n",
    "        << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    MutexLockGuard lock(mutex_); // 对比单线程\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      connections_.insert(conn);\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      connections_.erase(conn);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    MutexLockGuard lock(mutex_); // 对比单线程\n",
    "    for (ConnectionList::iterator it = connections_.begin();\n",
    "        it != connections_.end();\n",
    "        ++it)\n",
    "    {\n",
    "      codec_.send(get_pointer(*it), message);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  typedef std::set<TcpConnectionPtr> ConnectionList;\n",
    "  TcpServer server_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  MutexLock mutex_;\n",
    "  ConnectionList connections_ GUARDED_BY(mutex_);\n",
    "};\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 1)\n",
    "  {\n",
    "    EventLoop loop;\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));\n",
    "    InetAddress serverAddr(port);\n",
    "    ChatServer server(&loop, serverAddr);\n",
    "    if (argc > 2)\n",
    "    {\n",
    "      server.setThreadNum(atoi(argv[2]));  // 对比单线程\n",
    "    }\n",
    "    server.start();\n",
    "    loop.loop();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s port [thread_num]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 线程模型：\n",
    "  - 固定数量工作线程(one loop per thread)\n",
    "  - 新连接均匀分配到各线程\n",
    "- 消息广播：\n",
    "  - 简单遍历所有连接发送\n",
    "  - 使用全局锁保护共享资源\n",
    "- 优化方向：\n",
    "  - 减少临界区大小(版本C)\n",
    "  - 使用线程本地存储(版本D)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 聊天服务器代码改进 *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "聊天服务器代码改进-多线程版\n",
    "\n",
    "- 改进动机：单线程版本在多核机器上只能发挥一个核的作用，性能受限\n",
    "- 核心改动：\n",
    "  - 增加线程数设置：通过`server.setThreadNum()`指定工作线程数量\n",
    "  - 添加互斥锁保护：使用`MutexLock`保护共享的`connections_`集合\n",
    "- 实现特点：\n",
    "  - 每个线程运行独立的事件循环(EventLoop)\n",
    "  - 全局锁导致消息广播时存在线程竞争，性能提升有限"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "聊天服务器代码改进-高效版, \n",
    "\n",
    "- 优化技术：\n",
    "  - 使用共享指针管理连接集合：`typedef boost::shared_ptr<ConnectionList> ConnectionListPtr`\n",
    "  - 采用类似copy-on-write技术：获取连接列表时仅复制指针而非整个集合\n",
    "- 关键实现：\n",
    "  - `getConnectionList()`方法中加锁范围缩小到仅保护指针复制操作\n",
    "  - 遍历连接时使用局部副本，避免长时间持有锁\n",
    "- 性能优势：\n",
    "  - 临界区大幅缩小，仅与连接数变化频率相关\n",
    "  - 消息广播时不再阻塞其他线程操作"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/server_threaded_efficient.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/TcpServer.h\"\n",
    "\n",
    "#include <set>\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "class ChatServer : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatServer(EventLoop* loop,\n",
    "             const InetAddress& listenAddr)\n",
    "  : server_(loop, listenAddr, \"ChatServer\"),\n",
    "    codec_(std::bind(&ChatServer::onStringMessage, this, _1, _2, _3)),\n",
    "    connections_(new ConnectionList)\n",
    "  {\n",
    "    server_.setConnectionCallback(\n",
    "        std::bind(&ChatServer::onConnection, this, _1));\n",
    "    server_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "  }\n",
    "\n",
    "  void setThreadNum(int numThreads)\n",
    "  {\n",
    "    server_.setThreadNum(numThreads);\n",
    "  }\n",
    "\n",
    "  void start()\n",
    "  {\n",
    "    server_.start();\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->peerAddress().toIpPort() << \" -> \"\n",
    "        << conn->localAddress().toIpPort() << \" is \"\n",
    "        << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    MutexLockGuard lock(mutex_);\n",
    "    if (!connections_.unique())\n",
    "    {\n",
    "      connections_.reset(new ConnectionList(*connections_));\n",
    "    }\n",
    "    assert(connections_.unique());\n",
    "\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      connections_->insert(conn);\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      connections_->erase(conn);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  typedef std::set<TcpConnectionPtr> ConnectionList;\n",
    "  typedef std::shared_ptr<ConnectionList> ConnectionListPtr;\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    ConnectionListPtr connections = getConnectionList();\n",
    "    for (ConnectionList::iterator it = connections->begin();\n",
    "        it != connections->end();\n",
    "        ++it)\n",
    "    {\n",
    "      // 为了线程安全/跨线程使用TcpConnection, 会复制message, 因为无法保证调用send后, message仍然有效\n",
    "      // C++中没有像Java一样的引用技术, 调用函数传入参数, 为了在另一个线程再用这个参数, 需要拷贝\n",
    "      codec_.send(get_pointer(*it), message); \n",
    "    }\n",
    "  }\n",
    "\n",
    "  ConnectionListPtr getConnectionList()\n",
    "  {\n",
    "    MutexLockGuard lock(mutex_);\n",
    "    return connections_;\n",
    "  }\n",
    "\n",
    "  TcpServer server_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  MutexLock mutex_;\n",
    "  ConnectionListPtr connections_ GUARDED_BY(mutex_);\n",
    "};\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 1)\n",
    "  {\n",
    "    EventLoop loop;\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));\n",
    "    InetAddress serverAddr(port);\n",
    "    ChatServer server(&loop, serverAddr);\n",
    "    if (argc > 2)\n",
    "    {\n",
    "      server.setThreadNum(atoi(argv[2]));\n",
    "    }\n",
    "    server.start();\n",
    "    loop.loop();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s port [thread_num]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "聊天服务器代码改进-高性能版\n",
    "\n",
    "\n",
    "- 架构设计：\n",
    "  - 采用线程局部存储：`typedef ThreadLocalSingleton<ConnectionList> LocalConnections`\n",
    "  - 双重消息分发机制：先分发到各线程，再在线程内广播\n",
    "- 关键技术：\n",
    "  - 每个线程维护独立的连接集合\n",
    "  - 通过`queueInLoop`实现跨线程任务调度\n",
    "- 性能特点：\n",
    "  - 完全消除连接操作的锁竞争\n",
    "  - 数据拷贝次数仅与线程数相关，与连接数无关\n",
    "  - TCP连接不再跨线程调用，避免额外数据拷贝"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/server_threaded_highperformance.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/base/ThreadLocalSingleton.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/TcpServer.h\"\n",
    "\n",
    "#include <set>\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "class ChatServer : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatServer(EventLoop* loop,\n",
    "             const InetAddress& listenAddr)\n",
    "  : server_(loop, listenAddr, \"ChatServer\"),\n",
    "    codec_(std::bind(&ChatServer::onStringMessage, this, _1, _2, _3))\n",
    "  {\n",
    "    server_.setConnectionCallback(\n",
    "        std::bind(&ChatServer::onConnection, this, _1));\n",
    "    server_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "  }\n",
    "\n",
    "  void setThreadNum(int numThreads)\n",
    "  {\n",
    "    server_.setThreadNum(numThreads);\n",
    "  }\n",
    "\n",
    "  void start()\n",
    "  {\n",
    "    server_.setThreadInitCallback(std::bind(&ChatServer::threadInit, this, _1));\n",
    "    server_.start();\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->peerAddress().toIpPort() << \" -> \"\n",
    "             << conn->localAddress().toIpPort() << \" is \"\n",
    "             << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      LocalConnections::instance().insert(conn); // 局部线程对象, 不需要加锁\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      LocalConnections::instance().erase(conn);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    EventLoop::Functor f = std::bind(&ChatServer::distributeMessage, this, message);\n",
    "    LOG_DEBUG;\n",
    "\n",
    "    MutexLockGuard lock(mutex_); // 可以继续优化这个锁\n",
    "    for (std::set<EventLoop*>::iterator it = loops_.begin();\n",
    "        it != loops_.end();\n",
    "        ++it)\n",
    "    {\n",
    "      (*it)->queueInLoop(f); // loop会调用distributeMessage\n",
    "    }\n",
    "    LOG_DEBUG;\n",
    "  }\n",
    "\n",
    "  typedef std::set<TcpConnectionPtr> ConnectionList;\n",
    "\n",
    "  void distributeMessage(const string& message)\n",
    "  {\n",
    "    LOG_DEBUG << \"begin\";\n",
    "    for (ConnectionList::iterator it = LocalConnections::instance().begin();\n",
    "        it != LocalConnections::instance().end();\n",
    "        ++it)\n",
    "    {\n",
    "      codec_.send(get_pointer(*it), message);\n",
    "    }\n",
    "    LOG_DEBUG << \"end\";\n",
    "  }\n",
    "\n",
    "  void threadInit(EventLoop* loop)\n",
    "  {\n",
    "    assert(LocalConnections::pointer() == NULL);\n",
    "    LocalConnections::instance();\n",
    "    assert(LocalConnections::pointer() != NULL);\n",
    "    MutexLockGuard lock(mutex_);\n",
    "    loops_.insert(loop); // loops_对象跨线程\n",
    "  }\n",
    "\n",
    "  TcpServer server_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  typedef ThreadLocalSingleton<ConnectionList> LocalConnections;\n",
    "\n",
    "  MutexLock mutex_;\n",
    "  std::set<EventLoop*> loops_ GUARDED_BY(mutex_);\n",
    "};\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 1)\n",
    "  {\n",
    "    EventLoop loop;\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));\n",
    "    InetAddress serverAddr(port);\n",
    "    ChatServer server(&loop, serverAddr);\n",
    "    if (argc > 2)\n",
    "    {\n",
    "      server.setThreadNum(atoi(argv[2]));\n",
    "    }\n",
    "    server.start();\n",
    "    loop.loop();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s port [thread_num]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "优化措施\n",
    "\n",
    "- 进一步优化方向：\n",
    "  - 预构造广播消息缓冲区，避免重复编码\n",
    "  - 使用只读方式访问连接集合，完全消除临界区\n",
    "  - 优化`queueInLoop`内部的锁竞争\n",
    "- 实现建议：\n",
    "  - 对高频调用的`distributeMessage`进行性能剖析\n",
    "  - 考虑消息缓冲区池化技术减少内存分配开销"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "客户端代码\n",
    "\n",
    "- 设计特点：\n",
    "  - 独立线程处理网络I/O和用户输入\n",
    "  - 使用互斥锁保护跨线程共享的TcpConnection对象\n",
    "- 注意事项：\n",
    "  - 所有跨线程访问的连接操作必须加锁\n",
    "  - 可考虑单线程+事件驱动方式简化设计"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/client.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/net/EventLoopThread.h\"\n",
    "#include \"muduo/net/TcpClient.h\"\n",
    "\n",
    "#include <iostream>\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "class ChatClient : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatClient(EventLoop* loop, const InetAddress& serverAddr)\n",
    "    : client_(loop, serverAddr, \"ChatClient\"),\n",
    "      codec_(std::bind(&ChatClient::onStringMessage, this, _1, _2, _3))\n",
    "  {\n",
    "    client_.setConnectionCallback(\n",
    "        std::bind(&ChatClient::onConnection, this, _1));\n",
    "    client_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "    client_.enableRetry();\n",
    "  }\n",
    "\n",
    "  void connect()\n",
    "  {\n",
    "    client_.connect();\n",
    "  }\n",
    "\n",
    "  void disconnect()\n",
    "  {\n",
    "    client_.disconnect();\n",
    "  }\n",
    "\n",
    "  void write(const StringPiece& message)\n",
    "  {\n",
    "    MutexLockGuard lock(mutex_);\n",
    "    if (connection_)\n",
    "    {\n",
    "      codec_.send(get_pointer(connection_), message);\n",
    "    }\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->localAddress().toIpPort() << \" -> \"\n",
    "             << conn->peerAddress().toIpPort() << \" is \"\n",
    "             << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    MutexLockGuard lock(mutex_);\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      connection_ = conn;\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      connection_.reset();\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    printf(\"<<< %s\\n\", message.c_str());\n",
    "  }\n",
    "\n",
    "  TcpClient client_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  MutexLock mutex_;\n",
    "  TcpConnectionPtr connection_ GUARDED_BY(mutex_);\n",
    "};\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 2)\n",
    "  {\n",
    "    EventLoopThread loopThread;\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[2]));\n",
    "    InetAddress serverAddr(argv[1], port);\n",
    "\n",
    "    ChatClient client(loopThread.startLoop(), serverAddr);\n",
    "    client.connect();\n",
    "    std::string line;\n",
    "    while (std::getline(std::cin, line))\n",
    "    {\n",
    "      client.write(line);\n",
    "    }\n",
    "    client.disconnect();\n",
    "    CurrentThread::sleepUsec(1000*1000);  // wait for disconnect, see ace/logging/client.cc\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s host_ip port\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "负载测试\n",
    "\n",
    "- 测试方案：\n",
    "  - 批量创建客户端连接模拟高并发场景\n",
    "  - 统计消息往返时延分布情况\n",
    "- 关键指标：\n",
    "  - 连接建立成功率\n",
    "  - 消息传播的百分位延迟\n",
    "  - 系统吞吐量极限值"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/asio/chat/loadtest.cc\n",
    "\n",
    "#include \"examples/asio/chat/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Atomic.h\"\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/base/Mutex.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/EventLoopThreadPool.h\"\n",
    "#include \"muduo/net/TcpClient.h\"\n",
    "\n",
    "#include <stdio.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "int g_connections = 0;\n",
    "AtomicInt32 g_aliveConnections;\n",
    "AtomicInt32 g_messagesReceived;\n",
    "Timestamp g_startTime;\n",
    "std::vector<Timestamp> g_receiveTime;\n",
    "EventLoop* g_loop;\n",
    "std::function<void()> g_statistic;\n",
    "\n",
    "class ChatClient : noncopyable\n",
    "{\n",
    " public:\n",
    "  ChatClient(EventLoop* loop, const InetAddress& serverAddr)\n",
    "    : loop_(loop),\n",
    "      client_(loop, serverAddr, \"LoadTestClient\"),\n",
    "      codec_(std::bind(&ChatClient::onStringMessage, this, _1, _2, _3))\n",
    "  {\n",
    "    client_.setConnectionCallback(\n",
    "        std::bind(&ChatClient::onConnection, this, _1));\n",
    "    client_.setMessageCallback(\n",
    "        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));\n",
    "    //client_.enableRetry();\n",
    "  }\n",
    "\n",
    "  void connect()\n",
    "  {\n",
    "    client_.connect();\n",
    "  }\n",
    "\n",
    "  void disconnect()\n",
    "  {\n",
    "    // client_.disconnect();\n",
    "  }\n",
    "\n",
    "  Timestamp receiveTime() const { return receiveTime_; }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << conn->localAddress().toIpPort() << \" -> \"\n",
    "             << conn->peerAddress().toIpPort() << \" is \"\n",
    "             << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      connection_ = conn;\n",
    "      if (g_aliveConnections.incrementAndGet() == g_connections)\n",
    "      {\n",
    "        LOG_INFO << \"all connected\";\n",
    "        loop_->runAfter(10.0, std::bind(&ChatClient::send, this));\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      connection_.reset();\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onStringMessage(const TcpConnectionPtr&,\n",
    "                       const string& message,\n",
    "                       Timestamp)\n",
    "  {\n",
    "    // printf(\"<<< %s\\n\", message.c_str());\n",
    "    receiveTime_ = loop_->pollReturnTime();\n",
    "    int received = g_messagesReceived.incrementAndGet();\n",
    "    if (received == g_connections)\n",
    "    {\n",
    "      Timestamp endTime = Timestamp::now();\n",
    "      LOG_INFO << \"all received \" << g_connections << \" in \"\n",
    "               << timeDifference(endTime, g_startTime);\n",
    "      g_loop->queueInLoop(g_statistic);\n",
    "    }\n",
    "    else if (received % 1000 == 0)\n",
    "    {\n",
    "      LOG_DEBUG << received;\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void send()\n",
    "  {\n",
    "    g_startTime = Timestamp::now();\n",
    "    codec_.send(get_pointer(connection_), \"hello\");\n",
    "    LOG_DEBUG << \"sent\";\n",
    "  }\n",
    "\n",
    "  EventLoop* loop_;\n",
    "  TcpClient client_;\n",
    "  LengthHeaderCodec codec_;\n",
    "  TcpConnectionPtr connection_;\n",
    "  Timestamp receiveTime_;\n",
    "};\n",
    "\n",
    "void statistic(const std::vector<std::unique_ptr<ChatClient>>& clients)\n",
    "{\n",
    "  LOG_INFO << \"statistic \" << clients.size();\n",
    "  std::vector<double> seconds(clients.size());\n",
    "  for (size_t i = 0; i < clients.size(); ++i)\n",
    "  {\n",
    "    seconds[i] = timeDifference(clients[i]->receiveTime(), g_startTime);\n",
    "  }\n",
    "\n",
    "  std::sort(seconds.begin(), seconds.end());\n",
    "  for (size_t i = 0; i < clients.size(); i += std::max(static_cast<size_t>(1), clients.size()/20))\n",
    "  {\n",
    "    printf(\"%6zd%% %.6f\\n\", i*100/clients.size(), seconds[i]);\n",
    "  }\n",
    "  if (clients.size() >= 100)\n",
    "  {\n",
    "    printf(\"%6d%% %.6f\\n\", 99, seconds[clients.size() - clients.size()/100]);\n",
    "  }\n",
    "  printf(\"%6d%% %.6f\\n\", 100, seconds.back());\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  LOG_INFO << \"pid = \" << getpid();\n",
    "  if (argc > 3)\n",
    "  {\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[2]));\n",
    "    InetAddress serverAddr(argv[1], port);\n",
    "    g_connections = atoi(argv[3]);\n",
    "    int threads = 0;\n",
    "    if (argc > 4)\n",
    "    {\n",
    "      threads = atoi(argv[4]);\n",
    "    }\n",
    "\n",
    "    EventLoop loop;\n",
    "    g_loop = &loop;\n",
    "    EventLoopThreadPool loopPool(&loop, \"chat-loadtest\");\n",
    "    loopPool.setThreadNum(threads);\n",
    "    loopPool.start();\n",
    "\n",
    "    g_receiveTime.reserve(g_connections);\n",
    "    std::vector<std::unique_ptr<ChatClient>> clients(g_connections);\n",
    "    g_statistic = std::bind(statistic, std::ref(clients));\n",
    "\n",
    "    for (int i = 0; i < g_connections; ++i)\n",
    "    {\n",
    "      clients[i].reset(new ChatClient(loopPool.getNextLoop(), serverAddr));\n",
    "      clients[i]->connect();\n",
    "      usleep(200);\n",
    "    }\n",
    "\n",
    "    loop.loop();\n",
    "    // client.disconnect();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s host_ip port connections [threads]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# HUb server(pub/sub)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Topic based\n",
    "- Section 7.11 of my book\n",
    "- https://github.com/chenshuo/muduo/tree/master/examples/hub"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# HUB服务器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## codec.h/codec.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/codec.cc\n",
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/codec.h\n",
    "\n",
    "#include \"examples/hub/codec.h\"\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "using namespace pubsub;\n",
    "\n",
    "ParseResult pubsub::parseMessage(Buffer* buf,\n",
    "                                 string* cmd,\n",
    "                                 string* topic,\n",
    "                                 string* content)\n",
    "{\n",
    "  ParseResult result = kError;\n",
    "  const char* crlf = buf->findCRLF();\n",
    "  if (crlf)\n",
    "  {\n",
    "    const char* space = std::find(buf->peek(), crlf, ' ');\n",
    "    if (space != crlf)\n",
    "    {\n",
    "      cmd->assign(buf->peek(), space);\n",
    "      topic->assign(space+1, crlf);\n",
    "      if (*cmd == \"pub\")\n",
    "      {\n",
    "        const char* start = crlf + 2;\n",
    "        crlf = buf->findCRLF(start);\n",
    "        if (crlf)\n",
    "        {\n",
    "          content->assign(start, crlf);\n",
    "          buf->retrieveUntil(crlf+2);\n",
    "          result = kSuccess;\n",
    "        }\n",
    "        else\n",
    "        {\n",
    "          result = kContinue;\n",
    "        }\n",
    "      }\n",
    "      else\n",
    "      {\n",
    "        buf->retrieveUntil(crlf+2);\n",
    "        result = kSuccess;\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      result = kError;\n",
    "    }\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    result = kContinue;\n",
    "  }\n",
    "  return result;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 协议复杂度：比聊天服务器更复杂，支持多主题管理\n",
    "- 协议格式：\n",
    "  - 采用文本协议，命令与内容用空格分隔\n",
    "  - 每条消息以回车换行结束（\\r\\n）\n",
    "  - 主要命令：pub（发布）、sub（订阅）、unsub（取消订阅）\n",
    "- 消息解析：\n",
    "  - 先查找空格分隔命令和主题\n",
    "  - 对pub命令额外解析内容部分\n",
    "  - 错误处理：遇到协议错误会断开连接"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## pub.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/pub.cc\n",
    "\n",
    "#include \"examples/hub/pubsub.h\"\n",
    "#include \"muduo/base/ProcessInfo.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/EventLoopThread.h\"\n",
    "\n",
    "#include <iostream>\n",
    "#include <stdio.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "using namespace pubsub;\n",
    "\n",
    "EventLoop* g_loop = NULL;\n",
    "string g_topic;\n",
    "string g_content;\n",
    "\n",
    "void connection(PubSubClient* client)\n",
    "{\n",
    "  if (client->connected())\n",
    "  {\n",
    "    client->publish(g_topic, g_content);\n",
    "    client->stop();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    g_loop->quit();\n",
    "  }\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  if (argc == 4)\n",
    "  {\n",
    "    string hostport = argv[1];\n",
    "    size_t colon = hostport.find(':');\n",
    "    if (colon != string::npos)\n",
    "    {\n",
    "      string hostip = hostport.substr(0, colon);\n",
    "      uint16_t port = static_cast<uint16_t>(atoi(hostport.c_str()+colon+1));\n",
    "      g_topic = argv[2];\n",
    "      g_content = argv[3];\n",
    "\n",
    "      string name = ProcessInfo::username()+\"@\"+ProcessInfo::hostname();\n",
    "      name += \":\" + ProcessInfo::pidString();\n",
    "\n",
    "      if (g_content == \"-\")\n",
    "      {\n",
    "        EventLoopThread loopThread;\n",
    "        g_loop = loopThread.startLoop();\n",
    "        PubSubClient client(g_loop, InetAddress(hostip, port), name);\n",
    "        client.start();\n",
    "\n",
    "        string line;\n",
    "        while (getline(std::cin, line))\n",
    "        {\n",
    "          client.publish(g_topic, line);\n",
    "        }\n",
    "        client.stop();\n",
    "        CurrentThread::sleepUsec(1000*1000);\n",
    "      }\n",
    "      else\n",
    "      {\n",
    "        EventLoop loop;\n",
    "        g_loop = &loop;\n",
    "        PubSubClient client(g_loop, InetAddress(hostip, port), name);\n",
    "        client.setConnectionCallback(connection);\n",
    "        client.start();\n",
    "        loop.loop();\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      printf(\"Usage: %s hub_ip:port topic content\\n\", argv[0]);\n",
    "    }\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s hub_ip:port topic content\\n\"\n",
    "           \"Read contents from stdin:\\n\"\n",
    "           \"  %s hub_ip:port topic -\\n\", argv[0], argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心功能：\n",
    "  - 连接HUB服务器后立即发布指定主题内容\n",
    "  - 发布完成后自动断开连接\n",
    "- 使用场景：\n",
    "  - 命令行工具形式\n",
    "  - 适用于一次性发布场景"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## pubsub.h"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/pubsub.h\n",
    "\n",
    "#ifndef MUDUO_EXAMPLES_HUB_PUBSUB_H\n",
    "#define MUDUO_EXAMPLES_HUB_PUBSUB_H\n",
    "\n",
    "#include \"muduo/net/TcpClient.h\"\n",
    "\n",
    "namespace pubsub\n",
    "{\n",
    "using muduo::string;\n",
    "\n",
    "// FIXME: dtor is not thread safe\n",
    "class PubSubClient : muduo::noncopyable\n",
    "{\n",
    " public:\n",
    "  typedef std::function<void (PubSubClient*)> ConnectionCallback;\n",
    "  typedef std::function<void (const string& topic,\n",
    "                              const string& content,\n",
    "                              muduo::Timestamp)> SubscribeCallback;\n",
    "\n",
    "  PubSubClient(muduo::net::EventLoop* loop,\n",
    "               const muduo::net::InetAddress& hubAddr,\n",
    "               const string& name);\n",
    "  void start();\n",
    "  void stop();\n",
    "  bool connected() const;\n",
    "\n",
    "  void setConnectionCallback(const ConnectionCallback& cb)\n",
    "  { connectionCallback_ = cb; }\n",
    "\n",
    "  bool subscribe(const string& topic, const SubscribeCallback& cb);\n",
    "  void unsubscribe(const string& topic);\n",
    "  bool publish(const string& topic, const string& content);\n",
    "\n",
    " private:\n",
    "  void onConnection(const muduo::net::TcpConnectionPtr& conn);\n",
    "  void onMessage(const muduo::net::TcpConnectionPtr& conn,\n",
    "                 muduo::net::Buffer* buf,\n",
    "                 muduo::Timestamp receiveTime);\n",
    "  bool send(const string& message);\n",
    "\n",
    "  muduo::net::TcpClient client_;\n",
    "  muduo::net::TcpConnectionPtr conn_;\n",
    "  ConnectionCallback connectionCallback_;\n",
    "  SubscribeCallback subscribeCallback_;\n",
    "};\n",
    "}  // namespace pubsub\n",
    "\n",
    "#endif  // MUDUO_EXAMPLES_HUB_PUBSUB_H"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心接口：\n",
    "  - `publish()`：发布消息到指定主题\n",
    "  - `subscribe()`：订阅指定主题\n",
    "  - `unsubscribe()`：取消订阅\n",
    "- 回调机制：\n",
    "  - 连接状态回调（`ConnectionCallback`）\n",
    "  - 消息接收回调（`SubscribeCallback`）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## pubsub.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/pubsub.cc\n",
    "\n",
    "#include \"examples/hub/pubsub.h\"\n",
    "#include \"examples/hub/codec.h\"\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "using namespace pubsub;\n",
    "\n",
    "PubSubClient::PubSubClient(EventLoop* loop,\n",
    "                           const InetAddress& hubAddr,\n",
    "                           const string& name)\n",
    "  : client_(loop, hubAddr, name)\n",
    "{\n",
    "  // FIXME: dtor is not thread safe\n",
    "  client_.setConnectionCallback(\n",
    "      std::bind(&PubSubClient::onConnection, this, _1));\n",
    "  client_.setMessageCallback(\n",
    "      std::bind(&PubSubClient::onMessage, this, _1, _2, _3));\n",
    "}\n",
    "\n",
    "void PubSubClient::start()\n",
    "{\n",
    "  client_.connect();\n",
    "}\n",
    "\n",
    "void PubSubClient::stop()\n",
    "{\n",
    "  client_.disconnect();\n",
    "}\n",
    "\n",
    "bool PubSubClient::connected() const\n",
    "{\n",
    "  return conn_ && conn_->connected();\n",
    "}\n",
    "\n",
    "bool PubSubClient::subscribe(const string& topic, const SubscribeCallback& cb)\n",
    "{\n",
    "  string message = \"sub \" + topic + \"\\r\\n\";\n",
    "  subscribeCallback_ = cb;\n",
    "  return send(message);\n",
    "}\n",
    "\n",
    "void PubSubClient::unsubscribe(const string& topic)\n",
    "{\n",
    "  string message = \"unsub \" + topic + \"\\r\\n\";\n",
    "  send(message);\n",
    "}\n",
    "\n",
    "\n",
    "bool PubSubClient::publish(const string& topic, const string& content)\n",
    "{\n",
    "  string message = \"pub \" + topic + \"\\r\\n\" + content + \"\\r\\n\";\n",
    "  return send(message);\n",
    "}\n",
    "\n",
    "void PubSubClient::onConnection(const TcpConnectionPtr& conn)\n",
    "{\n",
    "  if (conn->connected())\n",
    "  {\n",
    "    conn_ = conn;\n",
    "    // FIXME: re-sub\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    conn_.reset();\n",
    "  }\n",
    "  if (connectionCallback_)\n",
    "  {\n",
    "    connectionCallback_(this);\n",
    "  }\n",
    "}\n",
    "\n",
    "void PubSubClient::onMessage(const TcpConnectionPtr& conn,\n",
    "                             Buffer* buf,\n",
    "                             Timestamp receiveTime)\n",
    "{\n",
    "  ParseResult result = kSuccess;\n",
    "  while (result == kSuccess)\n",
    "  {\n",
    "    string cmd;\n",
    "    string topic;\n",
    "    string content;\n",
    "    result = parseMessage(buf, &cmd, &topic, &content);\n",
    "    if (result == kSuccess)\n",
    "    {\n",
    "      if (cmd == \"pub\" && subscribeCallback_)\n",
    "      {\n",
    "        subscribeCallback_(topic, content, receiveTime);\n",
    "      }\n",
    "    }\n",
    "    else if (result == kError)\n",
    "    {\n",
    "      conn->shutdown();\n",
    "    }\n",
    "  }\n",
    "}\n",
    "\n",
    "bool PubSubClient::send(const string& message)\n",
    "{\n",
    "  bool succeed = false;\n",
    "  if (conn_ && conn_->connected())\n",
    "  {\n",
    "    conn_->send(message);\n",
    "    succeed = true;\n",
    "  }\n",
    "  return succeed;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 协议构造：\n",
    "  - pub命令格式：`pub <topic>\\r\\n<content>\\r\\n`\n",
    "  - sub命令格式：`sub <topic>\\r\\n`\n",
    "  - unsub命令格式：`unsub <topic>\\r\\n`\n",
    "- 消息处理：\n",
    "  - 使用parseMessage()解析服务器响应\n",
    "  - 对pub命令触发订阅回调"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## hub.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/hub.cc\n",
    "\n",
    "#include \"examples/hub/codec.h\"\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/TcpServer.h\"\n",
    "\n",
    "#include <map>\n",
    "#include <set>\n",
    "#include <stdio.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "namespace pubsub\n",
    "{\n",
    "\n",
    "typedef std::set<string> ConnectionSubscription;\n",
    "\n",
    "class Topic : public muduo::copyable\n",
    "{\n",
    " public:\n",
    "  Topic(const string& topic)\n",
    "    : topic_(topic)\n",
    "  {\n",
    "  }\n",
    "\n",
    "  void add(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    audiences_.insert(conn);\n",
    "    if (lastPubTime_.valid())\n",
    "    {\n",
    "      conn->send(makeMessage());\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void remove(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    audiences_.erase(conn);\n",
    "  }\n",
    "\n",
    "  void publish(const string& content, Timestamp time)\n",
    "  {\n",
    "    content_ = content;\n",
    "    lastPubTime_ = time;\n",
    "    string message = makeMessage();\n",
    "    for (std::set<TcpConnectionPtr>::iterator it = audiences_.begin();\n",
    "         it != audiences_.end();\n",
    "         ++it)\n",
    "    {\n",
    "      (*it)->send(message);\n",
    "    }\n",
    "  }\n",
    "\n",
    " private:\n",
    "\n",
    "  string makeMessage()\n",
    "  {\n",
    "    return \"pub \" + topic_ + \"\\r\\n\" + content_ + \"\\r\\n\";\n",
    "  }\n",
    "\n",
    "  string topic_;\n",
    "  string content_;\n",
    "  Timestamp lastPubTime_;\n",
    "  std::set<TcpConnectionPtr> audiences_;\n",
    "};\n",
    "\n",
    "class PubSubServer : noncopyable\n",
    "{\n",
    " public:\n",
    "  PubSubServer(muduo::net::EventLoop* loop,\n",
    "               const muduo::net::InetAddress& listenAddr)\n",
    "    : loop_(loop),\n",
    "      server_(loop, listenAddr, \"PubSubServer\")\n",
    "  {\n",
    "    server_.setConnectionCallback(\n",
    "        std::bind(&PubSubServer::onConnection, this, _1));\n",
    "    server_.setMessageCallback(\n",
    "        std::bind(&PubSubServer::onMessage, this, _1, _2, _3));\n",
    "    loop_->runEvery(1.0, std::bind(&PubSubServer::timePublish, this));\n",
    "  }\n",
    "\n",
    "  void start()\n",
    "  {\n",
    "    server_.start();\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void onConnection(const TcpConnectionPtr& conn)\n",
    "  {\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      conn->setContext(ConnectionSubscription());\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      const ConnectionSubscription& connSub\n",
    "        = boost::any_cast<const ConnectionSubscription&>(conn->getContext());\n",
    "      // subtle: doUnsubscribe will erase *it, so increase before calling.\n",
    "      for (ConnectionSubscription::const_iterator it = connSub.begin();\n",
    "           it != connSub.end();)\n",
    "      {\n",
    "        doUnsubscribe(conn, *it++);\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onMessage(const TcpConnectionPtr& conn,\n",
    "                 Buffer* buf,\n",
    "                 Timestamp receiveTime)\n",
    "  {\n",
    "    ParseResult result = kSuccess;\n",
    "    while (result == kSuccess)\n",
    "    {\n",
    "      string cmd;\n",
    "      string topic;\n",
    "      string content;\n",
    "      result = parseMessage(buf, &cmd, &topic, &content);\n",
    "      if (result == kSuccess)\n",
    "      {\n",
    "        if (cmd == \"pub\")\n",
    "        {\n",
    "          doPublish(conn->name(), topic, content, receiveTime);\n",
    "        }\n",
    "        else if (cmd == \"sub\")\n",
    "        {\n",
    "          LOG_INFO << conn->name() << \" subscribes \" << topic;\n",
    "          doSubscribe(conn, topic);\n",
    "        }\n",
    "        else if (cmd == \"unsub\")\n",
    "        {\n",
    "          doUnsubscribe(conn, topic);\n",
    "        }\n",
    "        else\n",
    "        {\n",
    "          conn->shutdown();\n",
    "          result = kError;\n",
    "        }\n",
    "      }\n",
    "      else if (result == kError)\n",
    "      {\n",
    "        conn->shutdown();\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void timePublish()\n",
    "  {\n",
    "    Timestamp now = Timestamp::now();\n",
    "    doPublish(\"internal\", \"utc_time\", now.toFormattedString(), now);\n",
    "  }\n",
    "\n",
    "  void doSubscribe(const TcpConnectionPtr& conn,\n",
    "                   const string& topic)\n",
    "  {\n",
    "    ConnectionSubscription* connSub\n",
    "      = boost::any_cast<ConnectionSubscription>(conn->getMutableContext());\n",
    "\n",
    "    connSub->insert(topic);\n",
    "    getTopic(topic).add(conn);\n",
    "  }\n",
    "\n",
    "  void doUnsubscribe(const TcpConnectionPtr& conn,\n",
    "                     const string& topic)\n",
    "  {\n",
    "    LOG_INFO << conn->name() << \" unsubscribes \" << topic;\n",
    "    getTopic(topic).remove(conn);\n",
    "    // topic could be the one to be destroyed, so don't use it after erasing.\n",
    "    ConnectionSubscription* connSub\n",
    "      = boost::any_cast<ConnectionSubscription>(conn->getMutableContext());\n",
    "    connSub->erase(topic);\n",
    "  }\n",
    "\n",
    "  void doPublish(const string& source,\n",
    "                 const string& topic,\n",
    "                 const string& content,\n",
    "                 Timestamp time)\n",
    "  {\n",
    "    getTopic(topic).publish(content, time);\n",
    "  }\n",
    "\n",
    "  Topic& getTopic(const string& topic)\n",
    "  {\n",
    "    std::map<string, Topic>::iterator it = topics_.find(topic);\n",
    "    if (it == topics_.end())\n",
    "    {\n",
    "      it = topics_.insert(make_pair(topic, Topic(topic))).first;\n",
    "    }\n",
    "    return it->second;\n",
    "  }\n",
    "\n",
    "  EventLoop* loop_;\n",
    "  TcpServer server_;\n",
    "  std::map<string, Topic> topics_;\n",
    "};\n",
    "\n",
    "}  // namespace pubsub\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  if (argc > 1)\n",
    "  {\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));\n",
    "    EventLoop loop;\n",
    "    if (argc > 2)\n",
    "    {\n",
    "      //int inspectPort = atoi(argv[2]);\n",
    "    }\n",
    "    pubsub::PubSubServer server(&loop, InetAddress(port));\n",
    "    server.start();\n",
    "    loop.loop();\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s pubsub_port [inspect_port]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 架构设计：\n",
    "  - 每个主题对应一个Topic对象\n",
    "  - 使用`std::set`管理订阅连接\n",
    "- 核心处理逻辑：\n",
    "  - 连接建立时初始化订阅上下文\n",
    "  - 连接断开时自动清理所有订阅\n",
    "  - 定时发布功能（timePublish）\n",
    "- 命令处理：\n",
    "  - pub：广播消息给所有订阅者\n",
    "  - sub：添加订阅关系\n",
    "  - unsub：移除订阅关系"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## sub.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/hub/sub.cc\n",
    "\n",
    "#include \"examples/hub/pubsub.h\"\n",
    "#include \"muduo/base/ProcessInfo.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "\n",
    "#include <vector>\n",
    "#include <stdio.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "using namespace pubsub;\n",
    "\n",
    "EventLoop* g_loop = NULL;\n",
    "std::vector<string> g_topics;\n",
    "\n",
    "void subscription(const string& topic, const string& content, Timestamp)\n",
    "{\n",
    "  printf(\"%s: %s\\n\", topic.c_str(), content.c_str());\n",
    "}\n",
    "\n",
    "void connection(PubSubClient* client)\n",
    "{\n",
    "  if (client->connected())\n",
    "  {\n",
    "    for (std::vector<string>::iterator it = g_topics.begin();\n",
    "        it != g_topics.end(); ++it)\n",
    "    {\n",
    "      client->subscribe(*it, subscription);\n",
    "    }\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    g_loop->quit();\n",
    "  }\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  if (argc > 2)\n",
    "  {\n",
    "    string hostport = argv[1];\n",
    "    size_t colon = hostport.find(':');\n",
    "    if (colon != string::npos)\n",
    "    {\n",
    "      string hostip = hostport.substr(0, colon);\n",
    "      uint16_t port = static_cast<uint16_t>(atoi(hostport.c_str()+colon+1));\n",
    "      for (int i = 2; i < argc; ++i)\n",
    "      {\n",
    "        g_topics.push_back(argv[i]);\n",
    "      }\n",
    "\n",
    "      EventLoop loop;\n",
    "      g_loop = &loop;\n",
    "      string name = ProcessInfo::username()+\"@\"+ProcessInfo::hostname();\n",
    "      name += \":\" + ProcessInfo::pidString();\n",
    "      PubSubClient client(&loop, InetAddress(hostip, port), name);\n",
    "      client.setConnectionCallback(connection);\n",
    "      client.start();\n",
    "      loop.loop();\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      printf(\"Usage: %s hub_ip:port topic [topic ...]\\n\", argv[0]);\n",
    "    }\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    printf(\"Usage: %s hub_ip:port topic [topic ...]\\n\", argv[0]);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 工作流程：\n",
    "  - 连接服务器后订阅指定主题\n",
    "  - 持续接收并打印消息\n",
    "- 消息处理：\n",
    "  - 简单打印格式：`<topic>: <content>`\n",
    "- 使用方式：\n",
    "  - 命令行参数指定服务器和主题\n",
    "  - 支持同时订阅多个主题"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 设计难点"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 实现初衷: 实现消息订阅机制是muduo网络库最早的开发动机之一\n",
    "- 功能特点: 采用Topic-based发布/订阅模式，参考书籍第7.11章节内容\n",
    "- 复杂度说明: 实际应用中可扩展安全性和可靠性措施，已有成熟中间件实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接收端接收慢导致的数据堆积问题\n",
    "\n",
    "- 现象表现: 当部分客户端接收速度慢时，数据会堆积在服务器内存中\n",
    "- 实验验证: 可通过`ctrl+z`暂停客户端观察服务器内存增长情况\n",
    "- 底层机制: muduo采用非阻塞发送，使用用户态缓冲区暂存待发数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "echo服务器限流方法\n",
    "- 同步机制: 在echo服务中，收发使用同一连接，可简单实现流量控制\n",
    "- 实现原理: 若客户端只接收500k/1m数据，服务器将暂停接收新数据直到积压数据被处理\n",
    "- 技术术语: 这种机制称为SWAT（Send-Wait-Acknowledge-Transfer）限流"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "聊天服务器限流问题及解决方案\n",
    "\n",
    "- 问题特殊性: 一对多场景下，单个慢客户端不应影响其他正常客户端\n",
    "- 拒绝服务风险: 恶意客户端可通过故意接收缓慢导致系统瘫痪\n",
    "- 解决方案:\n",
    "  - 协议优化: 接收端反馈进度，合并后续数据包（如3个snapshot合并发送）\n",
    "  - 磁盘卸载: 将积压数据写入磁盘文件释放内存（offload技术）\n",
    "  - 强制断开: 对长期慢速客户端实施踢除策略\n",
    "  - 业务妥协: 比分转播等场景可只发送最新状态，忽略中间过程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "一对多发送场景下的接收慢问题 \n",
    "- 核心原则: 单个慢客户端不应拖垮整个系统\n",
    "- 带宽适配: 针对低带宽客户端可调整发送策略（如降低更新频率）\n",
    "- 业务决策: 最终解决方案需结合具体业务场景制定\n",
    "- 技术边界: 纯网络层无法解决的问题需要业务逻辑配合处理"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TCP relay功能描述及Python实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Contents"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP relay in Python, Muduo and Go\n",
    "- Socks4a proxy server: muduo/examples/socks4a\n",
    "- Alternative approach:NAT\n",
    "- TCP is reliable, but not infallible\n",
    "- Muduo in C++11/14/17"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP Relay, simple at first glance"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- For each client connection `S`, create a connection `C` to server, then forward data between `S` and `C` bi-directionally.\n",
    "\n",
    "![8.4](./images/8.4.png)\n",
    "\n",
    "- §7.13 of the muduobook, with updated/corrected code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 基本架构: 服务器在2000端口侦听，客户端连接后建立到3000端口服务器的连接\n",
    "- 数据转发: 建立双向连接后，在客户端连接(S)和服务器连接(C)之间双向转发数据\n",
    "- 带宽适配: 需要处理两端带宽不匹配的情况(如千兆网与ADSL两兆连接)\n",
    "- 代码演进: 相比书中第7.13节示例，更新了处理带宽不匹配的代码实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "TCP relay in Python\n",
    "\n",
    "- 实现位置: 代码位于[recipes/python/tcprelay2.py](https://github.com/chenshuo/recipes/blob/master/python/tcprelay2.py)\n",
    "- 简化版本: 相比书中版本移除了延时处理，专注于核心功能\n",
    "- 建立流程:\n",
    "  - 创建TCP socket\n",
    "  - 设置`SO_REUSEADDR`选项\n",
    "  - 绑定(bind)和监听(listen)\n",
    "  - 循环接受(accept)连接\n",
    "- 线程模型: 为每个连接创建两个线程处理双向数据转发\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP relay in Python: forward() function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "python"
    }
   },
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "\n",
    "import socket, thread, time\n",
    "\n",
    "def forward(source, destination):\n",
    "    source_addr = source.getpeername()\n",
    "    while True:  # FIXME: error handling\n",
    "        data = source.recv(4096)  # Connection reset by peer\n",
    "        if data:\n",
    "            destination.sendall(data)  # Broken pipe\n",
    "        else:\n",
    "            print 'disconnect', source_addr\n",
    "            destination.shutdown(socket.SHUT_WR) # Will talk TCP halfclose later\n",
    "            break\n",
    "\n",
    "# 建立TCP服务器\n",
    "listensocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) \n",
    "listensocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n",
    "listensocket.bind(('', 2000))\n",
    "listensocket.listen(5)\n",
    "\n",
    "while True:\n",
    "    (serversocket, address) = listensocket.accept()\n",
    "    print 'accepted', address\n",
    "    clientsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "    clientsocket.connect(('localhost', 3000))\n",
    "    print 'connected', clientsocket.getpeername()\n",
    "    thread.start_new_thread(forward, (serversocket, clientsocket)) # 启动线程, forward\n",
    "    thread.start_new_thread(forward, (clientsocket, serversocket))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Bandwidth mismatch? Slow writing throttles reading\n",
    "- !!! No proper close of connections\n",
    "- !!! No error handling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 核心逻辑: 在死循环中从`source`读取4K数据并转发到`destination`\n",
    "- 阻塞特性: 使用阻塞IO自然解决带宽不匹配问题(发送慢则读取也变慢)\n",
    "- 关闭处理: 使用`shutdown(SHUT_WR)`而非`close`，涉及TCP半关闭概念\n",
    "- 错误处理缺陷:\n",
    "  - 未处理`sendall`可能抛出的`Broken pipe`异常\n",
    "  - 未处理`recv`可能抛出的`Connection reset by peer`异常\n",
    "- 数据流控制: 当`destination`接收慢时，会阻塞`sender`线程，进而减缓`source`的读取速度"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TCP half-closed connection"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Not widely used, except for transparent proxy\n",
    "  - https://www.zhihu.com/question/48871684/answer/113135138\n",
    "- Muduo doesn’t fully support it (§7.2p.181)\n",
    "  - `TcpConnection::shutdown()` => `shutdown(sockfd, SHUT_WR)`\n",
    "  - `Keep read(sockfd)` until it returns `0`, then `close(sockfd)`\n",
    "  - !!! Muduo doesn’t \n",
    "- Netty supports it since 4.0(released in 2013/07)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 基本概念：TCP是双向协议，允许单向关闭形成半关连接状态（只发送不接收或只接收不发送）\n",
    "- 应用场景：主要用于透明代理实现，普通网络协议通常直接完全关闭连接\n",
    "- Muduo实现特点：\n",
    "  - 仅支持部分半关功能（$7.2 p.181$﻿）\n",
    "  - 调用`shutdown(sockfd, SHUT_WR)`关闭写端\n",
    "  - 持续`read(sockfd)`直到返回`0`才关闭连接\n",
    "  - 不支持在对方关闭写端后继续发送数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 作者：胡宇光\n",
    "链接：https://www.zhihu.com/question/48871684/answer/113135138\n",
    "来源：知乎\n",
    "著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。<br/>\n",
    "> `shutdown(both)`还需要`close`，不然你就泄漏了一个或几个`handle`或`fd`以及相关资源。`read`返回`0`表示你收到了对方发来的`fin`，这可能是对方调用`shutdown(send)`，也可能是对方调用了`close()`。从tcp协议本身来说，你是无法知道对方到底是调用了`close()`还是调用了`shutdown(send)`的，os的tcp协议栈也不知道。因此此时是否要`close`取决于你的应用。通常来说如果对方调用的是`close`，那么你也可以`close`。否则你不能`close`，例如对方发送一个包给你并`shutdown write`然后调用`recv`，这时候你还可以返回一个或多个包，连接此时处于半关闭状态，可以一直持续。这么做的客户端不多（`connect, send, shtudown(send), recv();`），但的确有，而且是完全合法的。如果通讯双方都是你自己的代码，那么你知道是哪种情况。如果你不能了解对方的代码，甚至你是个proxy，两边代码你都不了解，那么通常来说你不能`close`。<br/>\n",
    "很多server/proxy的实现为当`read`返回`0`就`close`，这种实现是错误的，这个实现无法兼容刚才我说的那种情况。对于proxy来说，正确的做法是透传双方的行为。因此，当你`read(client_side_socket)`返回`0`时，你应该对另外一端调用`shutdown(server_side_socket, send)`，这样服务器就会`read`返回`0`，你透明的传递了这个行为。那么作为proxy，你什么时候才能`close`呢？`client_socket`和`server_socket`上`read`都返回了`0`，或者有任何一方返回了`-1`时你可以`close`。当然你也可以考虑设置一个超时时间，如果线路上超过5分钟没有数据你就断开，但这是另一个维度的问题。<br/>\n",
    "关于`close`，要注意的是默认情况下它是一个异步的过程。作为proxy来说，如果你想避免大量`close_wait`，那么你可以在`close`之前`shutdown`，然后启动一个5s的`delaytask`，在`delaytask`里设置超时时间为`0`的`so_linger`，然后`close`，对`socket`进行`hard close`。这时候`close`是同步的，如果此时不能优雅关闭，那么系统会立刻强制关闭。<br/>\n",
    "我没有看muduo的代码。如果你在2处`close`同一个socket，这是高危行为。因为2次`close`之间很可能会有一个新socket产生并且值和你第一次`close`的那个一样。你第二次`close`就会错误的`close`了一个不属于你的socket。这种错误非常难查，因此绝对不要干这种事。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "知乎答案分析\n",
    "\n",
    "- 核心问题：\n",
    "  - 无法通过协议栈判断对方调用的是`close()`还是`shutdown(send)`\n",
    "  - `read`返回`0`可能是收到`FIN`（对方`shutdown(send)`或`close()`）\n",
    "- 正确处理逻辑：\n",
    "  - 对方调用`close`时可安全关闭\n",
    "  - 对方`shutdown write`后若仍需返回数据，应保持半关状态\n",
    "  - Proxy需透传双方行为：当`read(client)`返回`0`时应对服务端调用`shutdown(server_side_socket,send)`\n",
    "- Proxy实现要点：\n",
    "  - 关闭时机：两端`read`都返回`0`或任意端返回`-1`\n",
    "  - 可设置超时机制（如5分钟无数据断开）\n",
    "  - 避免大量`CLOSE_WAIT`：先`shutdown`再设置`SO_LINGER`硬关闭\n",
    "\n",
    "![8.5](./images/8.5.png) <br/>\n",
    "\n",
    "![8.6](./images/8.6.png) <br/>\n",
    "\n",
    "- Muduo设计考量：\n",
    "  - 状态图简化（图8-6）直接进入`disconnected`状态\n",
    "  - 不支持`read`返回`0`后继续发送的中间状态\n",
    "  - 与Netty对比：Netty 4.0+（2013/07）才支持完整半关功能\n",
    "- 关键状态解析：\n",
    "  - `CLOSE_WAIT`：可继续发送数据的半关状态\n",
    "  - `FIN_WAIT_2`：主动关闭方等待最终`ACK`的状态\n",
    "  - 典型场景：调用`shutdown(SHUT_WR)`会立即发送`FIN`，快速进入`FIN_WAIT_1→FIN_WAIT_2`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 非阻塞TCP relay实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Non-blocking TCP relay is awful"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- An analogy of water bucket"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "有两个连接的情况\n",
    "\n",
    "- 基本实现原理：在非阻塞IO下实现TCP relay时，若有两个连接（如s连接和c连接），只需在各自的`on_message`回调中将接收到的数据通过对方的`send`方法转发即可。\n",
    "- 单线程优势：这种实现不需要两个线程，因为非阻塞IO基于事件驱动，单个线程就能处理两个连接的消息到达事件。\n",
    "- 内存风险：当一端发送速度远快于另一端接收速度时，未发送数据会持续累积在内存中，导致内存暴涨。例如c连接快速发送而s连接接收缓慢时，c的`send`缓存会不断增长"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "水桶比喻法\n",
    "\n",
    "- 无限缓存问题：原始实现相当于一个深度无限的水桶，入水口(read)持续接收数据，出水口(send)速度不足时会导致水位(内存占用)无限上涨。\n",
    "- 流量控制需求：需要像真实水桶那样设置容量限制，当水位达到警戒线时关闭入水口，防止溢出。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "高水位回调\n",
    "\n",
    "- high water mark机制：设置高水位阈值（如1MB），当待发送数据超过该阈值时触发`high water mark callback`。\n",
    "- 流量控制操作：在回调中执行`stop_read`操作，停止接收新数据但保持发送，此时内存占用将稳定在高水位线。\n",
    "- `write complete callback`：当所有积压数据发送完毕（水位降为零）时触发，此时应重新`start_read`恢复数据接收。\n",
    "- 动态调节效果：这种机制形成\"停止接收-清空缓存-恢复接收\"的循环，相比阻塞IO需要更多步骤但能有效控制内存使用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 源码及运行"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- https://github.com/chenshuo/muduo/tree/master/examples/socks4a"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "主要文件结构\n",
    "\n",
    "- 核心文件：\n",
    "  - `tunnel.h`：实现两个连接间的数据交换核心逻辑\n",
    "  - `tcprelay.cc`：主程序文件\n",
    "  - `socks4a.cc`：TCP中继的升级版本\n",
    "- 实现特点：三个独立main函数共享tunnel核心功能"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/socks4a/tcprelay.cc\n",
    "\n",
    "#include \"examples/socks4a/tunnel.h\"\n",
    "\n",
    "#include <malloc.h>\n",
    "#include <stdio.h>\n",
    "#include <sys/resource.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "EventLoop* g_eventLoop;\n",
    "InetAddress* g_serverAddr;\n",
    "std::map<string, TunnelPtr> g_tunnels;\n",
    "\n",
    "void onServerConnection(const TcpConnectionPtr& conn)\n",
    "{\n",
    "  LOG_DEBUG << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "  if (conn->connected())\n",
    "  {\n",
    "    conn->setTcpNoDelay(true);\n",
    "    conn->stopRead(); // 可以尝试注释, 看是否发生堆积\n",
    "    TunnelPtr tunnel(new Tunnel(g_eventLoop, *g_serverAddr, conn));\n",
    "    tunnel->setup();\n",
    "    tunnel->connect();\n",
    "    g_tunnels[conn->name()] = tunnel;\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    assert(g_tunnels.find(conn->name()) != g_tunnels.end());\n",
    "    g_tunnels[conn->name()]->disconnect();\n",
    "    g_tunnels.erase(conn->name());\n",
    "  }\n",
    "}\n",
    "\n",
    "void onServerMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp)\n",
    "{\n",
    "  LOG_DEBUG << buf->readableBytes();\n",
    "  if (!conn->getContext().empty())\n",
    "  {\n",
    "    const TcpConnectionPtr& clientConn\n",
    "      = boost::any_cast<const TcpConnectionPtr&>(conn->getContext());\n",
    "    clientConn->send(buf);\n",
    "  }\n",
    "}\n",
    "\n",
    "void memstat()\n",
    "{\n",
    "  malloc_stats();\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  if (argc < 4)\n",
    "  {\n",
    "    fprintf(stderr, \"Usage: %s <host_ip> <port> <listen_port>\\n\", argv[0]);\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    LOG_INFO << \"pid = \" << getpid() << \", tid = \" << CurrentThread::tid();\n",
    "    {\n",
    "      // set max virtual memory to 256MB.\n",
    "      size_t kOneMB = 1024*1024;\n",
    "      rlimit rl = { 256*kOneMB, 256*kOneMB };\n",
    "      setrlimit(RLIMIT_AS, &rl);\n",
    "    }\n",
    "    const char* ip = argv[1];\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[2]));\n",
    "    InetAddress serverAddr(ip, port);\n",
    "    g_serverAddr = &serverAddr;\n",
    "\n",
    "    uint16_t acceptPort = static_cast<uint16_t>(atoi(argv[3]));\n",
    "    InetAddress listenAddr(acceptPort);\n",
    "\n",
    "    EventLoop loop;\n",
    "    g_eventLoop = &loop;\n",
    "    loop.runEvery(3, memstat);\n",
    "\n",
    "    TcpServer server(&loop, listenAddr, \"TcpRelay\");\n",
    "\n",
    "    server.setConnectionCallback(onServerConnection);\n",
    "    server.setMessageCallback(onServerMessage);\n",
    "\n",
    "    server.start();\n",
    "\n",
    "    loop.loop();\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "tcprelay.cc, 主程序初始化\n",
    "\n",
    "- 内存控制：\n",
    "  - 显式限制虚拟内存为256MB（256*1024*1024字节）\n",
    "  - 防止非阻塞实现不当导致内存爆炸\n",
    "  - 通过setrlimit(RLIMIT_AS, &rl)系统调用实现\n",
    "- 参数处理：\n",
    "  - 必需参数：`<host_ip> <port> <listen_port>`\n",
    "  - 端口转换使用`static_cast<uint16_t>(atoi())`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "tcprelay.cc, TCPServer配置\n",
    "\n",
    "- 事件循环：\n",
    "  - 创建`EventLoop`对象并设为全局变量\n",
    "  - 每3秒执行`memstat()`内存统计\n",
    "- 回调机制：\n",
    "  - `setConnectionCallback`：处理连接建立/断开\n",
    "  - `setMessageCallback`：处理接收数据\n",
    "  - 使用`TcpServer`在指定端口监听"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "tcprelay.cc, 连接管理逻辑\n",
    "\n",
    "- 连接建立：\n",
    "  - 设置`TCP_NODELAY`禁用Nagle算法\n",
    "  - 立即调用`stopRead()`防止数据堆积\n",
    "  - 创建`Tunnel`对象并建立到目标服务器的连接\n",
    "- 连接断开：\n",
    "  - 从全局`g_tunnels`映射中移除对应隧道\n",
    "  - 调用`disconnect()`清理资源\n",
    "- 术语注意：代码中\"server\"实际指代客户端连接，需注意命名反直觉"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/socks4a/tunnel.h\n",
    "\n",
    "#ifndef MUDUO_EXAMPLES_SOCKS4A_TUNNEL_H\n",
    "#define MUDUO_EXAMPLES_SOCKS4A_TUNNEL_H\n",
    "\n",
    "#include \"muduo/base/Logging.h\"\n",
    "#include \"muduo/net/EventLoop.h\"\n",
    "#include \"muduo/net/InetAddress.h\"\n",
    "#include \"muduo/net/TcpClient.h\"\n",
    "#include \"muduo/net/TcpServer.h\"\n",
    "\n",
    "class Tunnel : public std::enable_shared_from_this<Tunnel>,\n",
    "               muduo::noncopyable\n",
    "{\n",
    " public:\n",
    "  Tunnel(muduo::net::EventLoop* loop,\n",
    "         const muduo::net::InetAddress& serverAddr,\n",
    "         const muduo::net::TcpConnectionPtr& serverConn)\n",
    "    : client_(loop, serverAddr, serverConn->name()),\n",
    "      serverConn_(serverConn)\n",
    "  {\n",
    "    LOG_INFO << \"Tunnel \" << serverConn->peerAddress().toIpPort()\n",
    "             << \" <-> \" << serverAddr.toIpPort();\n",
    "  }\n",
    "\n",
    "  ~Tunnel()\n",
    "  {\n",
    "    LOG_INFO << \"~Tunnel\";\n",
    "  }\n",
    "\n",
    "  void setup()\n",
    "  {\n",
    "    using std::placeholders::_1;\n",
    "    using std::placeholders::_2;\n",
    "    using std::placeholders::_3;\n",
    "\n",
    "    client_.setConnectionCallback(\n",
    "        std::bind(&Tunnel::onClientConnection, shared_from_this(), _1));\n",
    "    client_.setMessageCallback(\n",
    "        std::bind(&Tunnel::onClientMessage, shared_from_this(), _1, _2, _3));\n",
    "    serverConn_->setHighWaterMarkCallback(\n",
    "        std::bind(&Tunnel::onHighWaterMarkWeak,\n",
    "                  std::weak_ptr<Tunnel>(shared_from_this()), kServer, _1, _2),\n",
    "        1024*1024);\n",
    "  }\n",
    "\n",
    "  void connect()\n",
    "  {\n",
    "    client_.connect();\n",
    "  }\n",
    "\n",
    "  void disconnect()\n",
    "  {\n",
    "    client_.disconnect();\n",
    "    // serverConn_.reset();\n",
    "  }\n",
    "\n",
    " private:\n",
    "  void teardown()\n",
    "  {\n",
    "    client_.setConnectionCallback(muduo::net::defaultConnectionCallback);\n",
    "    client_.setMessageCallback(muduo::net::defaultMessageCallback);\n",
    "    if (serverConn_)\n",
    "    {\n",
    "      serverConn_->setContext(boost::any());\n",
    "      serverConn_->shutdown();\n",
    "    }\n",
    "    clientConn_.reset();\n",
    "  }\n",
    "\n",
    "  void onClientConnection(const muduo::net::TcpConnectionPtr& conn)\n",
    "  {\n",
    "    using std::placeholders::_1;\n",
    "    using std::placeholders::_2;\n",
    "\n",
    "    LOG_DEBUG << (conn->connected() ? \"UP\" : \"DOWN\");\n",
    "    if (conn->connected())\n",
    "    {\n",
    "      conn->setTcpNoDelay(true);\n",
    "      conn->setHighWaterMarkCallback(\n",
    "          std::bind(&Tunnel::onHighWaterMarkWeak,\n",
    "                    std::weak_ptr<Tunnel>(shared_from_this()), kClient, _1, _2),\n",
    "          1024*1024);\n",
    "      serverConn_->setContext(conn);\n",
    "      serverConn_->startRead();\n",
    "      clientConn_ = conn;\n",
    "      if (serverConn_->inputBuffer()->readableBytes() > 0)\n",
    "      {\n",
    "        conn->send(serverConn_->inputBuffer()); // 操作会清空原缓冲区（所有权转移）\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      teardown();\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onClientMessage(const muduo::net::TcpConnectionPtr& conn,\n",
    "                       muduo::net::Buffer* buf,\n",
    "                       muduo::Timestamp)\n",
    "  {\n",
    "    LOG_DEBUG << conn->name() << \" \" << buf->readableBytes();\n",
    "    if (serverConn_)\n",
    "    {\n",
    "      serverConn_->send(buf);\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      buf->retrieveAll();\n",
    "      abort();\n",
    "    }\n",
    "  }\n",
    "\n",
    "  enum ServerClient\n",
    "  {\n",
    "    kServer, kClient\n",
    "  };\n",
    "\n",
    "  // 高低水位回调\n",
    "  void onHighWaterMark(ServerClient which,\n",
    "                       const muduo::net::TcpConnectionPtr& conn,\n",
    "                       size_t bytesToSent)\n",
    "  {\n",
    "    using std::placeholders::_1;\n",
    "\n",
    "    LOG_INFO << (which == kServer ? \"server\" : \"client\")\n",
    "             << \" onHighWaterMark \" << conn->name()\n",
    "             << \" bytes \" << bytesToSent;\n",
    "\n",
    "    if (which == kServer) // serverconnection\n",
    "    {\n",
    "      if (serverConn_->outputBuffer()->readableBytes() > 0) // 避免竞态条件\n",
    "      {\n",
    "        clientConn_->stopRead(); // clientconnection停止读\n",
    "        serverConn_->setWriteCompleteCallback(\n",
    "            std::bind(&Tunnel::onWriteCompleteWeak,\n",
    "                      std::weak_ptr<Tunnel>(shared_from_this()), kServer, _1));\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      if (clientConn_->outputBuffer()->readableBytes() > 0)\n",
    "      {\n",
    "        serverConn_->stopRead();\n",
    "        clientConn_->setWriteCompleteCallback(\n",
    "            std::bind(&Tunnel::onWriteCompleteWeak,\n",
    "                      std::weak_ptr<Tunnel>(shared_from_this()), kClient, _1));\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "\n",
    "  static void onHighWaterMarkWeak(const std::weak_ptr<Tunnel>& wkTunnel,\n",
    "                                  ServerClient which,\n",
    "                                  const muduo::net::TcpConnectionPtr& conn,\n",
    "                                  size_t bytesToSent)\n",
    "  {\n",
    "    std::shared_ptr<Tunnel> tunnel = wkTunnel.lock();\n",
    "    if (tunnel)\n",
    "    {\n",
    "      tunnel->onHighWaterMark(which, conn, bytesToSent);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  void onWriteComplete(ServerClient which, const muduo::net::TcpConnectionPtr& conn)\n",
    "  {\n",
    "    LOG_INFO << (which == kServer ? \"server\" : \"client\")\n",
    "             << \" onWriteComplete \" << conn->name();\n",
    "    if (which == kServer)\n",
    "    {\n",
    "      clientConn_->startRead();\n",
    "      serverConn_->setWriteCompleteCallback(muduo::net::WriteCompleteCallback());\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      serverConn_->startRead();\n",
    "      clientConn_->setWriteCompleteCallback(muduo::net::WriteCompleteCallback());\n",
    "    }\n",
    "  }\n",
    "\n",
    "  static void onWriteCompleteWeak(const std::weak_ptr<Tunnel>& wkTunnel,\n",
    "                                  ServerClient which,\n",
    "                                  const muduo::net::TcpConnectionPtr& conn)\n",
    "  {\n",
    "    std::shared_ptr<Tunnel> tunnel = wkTunnel.lock();\n",
    "    if (tunnel)\n",
    "    {\n",
    "      tunnel->onWriteComplete(which, conn);\n",
    "    }\n",
    "  }\n",
    "\n",
    " private:\n",
    "  muduo::net::TcpClient client_;\n",
    "  muduo::net::TcpConnectionPtr serverConn_;\n",
    "  muduo::net::TcpConnectionPtr clientConn_;\n",
    "};\n",
    "typedef std::shared_ptr<Tunnel> TunnelPtr;\n",
    "\n",
    "#endif  // MUDUO_EXAMPLES_SOCKS4A_TUNNEL_H"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "消息转发机制\n",
    "\n",
    "- 数据流向：\n",
    "  - 服务端消息：通过`getContext()`获取客户端连接后直接转发\n",
    "  - 客户端消息：对称转发至服务端连接\n",
    "- 缓冲区管理：\n",
    "  - 使用`Buffer`对象进行数据中转\n",
    "  - `send()`操作会清空原缓冲区（所有权转移）\n",
    "- 异常处理：当连接异常时调用`retrieveAll()`清空缓冲"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "流量控制实现-高低水位回调\n",
    "\n",
    "- 触发条件：\n",
    "  - 默认高水位阈值设置为1MB（1024*1024字节）\n",
    "  - 当输出缓冲区超过阈值时触发\n",
    "- 控制策略：\n",
    "  - 服务端高水位：停止客户端读取，注册写完成回调\n",
    "  - 客户端高水位：对称处理服务端连接\n",
    "- 关键处理：必须检查对端输出缓冲区是否为空（避免竞态条件）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "流量控制实现-写完成回调\n",
    "\n",
    "- 恢复机制：\n",
    "  - 数据发送完成后重新启用对端读取\n",
    "  - 清除写完成回调避免重复触发\n",
    "- 弱回调设计：\n",
    "  - 使用`weak_ptr`避免循环引用\n",
    "  - 通过`lock()`检查对象有效性"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Complex object lifetime management even with smart pointers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实际测试验证"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 基础功能测试\n",
    "\n",
    "- TCP Replay\n",
    "\n",
    "```shell\n",
    "bin/tcpreplay 127.0.0.1 3000 2000 # 监听2000, 数据发送到3000\n",
    "```\n",
    "\n",
    "- Server\n",
    "\n",
    "```shell\n",
    "nc -l -p 3000 # 监听3000\n",
    "```\n",
    "\n",
    "- Client\n",
    "\n",
    "```shell\n",
    "nc localhost 2000\n",
    "```\n",
    "\n",
    "验证结果：双向数据转发功能正常，延迟可接受"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 断连重试测试"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 重试策略：\n",
    "  - 初始重试间隔500ms\n",
    "  - 采用指数退避策略（500ms→1s→2s→4s→8s）\n",
    "  - 最大重试间隔8秒\n",
    "- 数据缓存：先停服务端, 然后客户端数据在服务端不可达时仍可发送，建立连接后自动转发, 客户端收到了AC, 并不代表数据到了服务端"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 高负载测试 \n",
    "\n",
    "- TCP Replay\n",
    "\n",
    "```shell\n",
    "bin/tcpreplay 127.0.0.1 3000 2000 # 监听2000, 数据发送到3000\n",
    "```\n",
    "\n",
    "- Server\n",
    "\n",
    "```shell\n",
    "nc -l -p 3000 > /dev/null # 监听3000\n",
    "```\n",
    "\n",
    "- Client\n",
    "\n",
    "```shell\n",
    "pv /dev/zero | nc localhost 2000\n",
    "```\n",
    "\n",
    "```shell\n",
    "utop\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 吞吐性能：\n",
    "  - 无限制时可达450MB/s\n",
    "  - CPU利用率接近100%（双核机器）\n",
    "- 内存控制：\n",
    "  - 实际内存使用维持在几MB级别\n",
    "  - 高低水位机制有效防止内存爆炸"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 限速测试\n",
    "\n",
    "- TCP Replay\n",
    "\n",
    "```shell\n",
    "bin/tcpreplay 127.0.0.1 3000 2000 # 监听2000, 数据发送到3000\n",
    "```\n",
    "\n",
    "- Server\n",
    "\n",
    "```shell\n",
    "nc -l -p 3000 | pv > /dev/null # 监听3000\n",
    "```\n",
    "\n",
    "```shell\n",
    "nc -l -p 3000 | pv -L 1m > /dev/null # 监听3000\n",
    "```\n",
    "\n",
    "- Client\n",
    "\n",
    "```shell\n",
    "nc localhost 2000 < /dev/zero\n",
    "```\n",
    "\n",
    "```shell\n",
    "utop\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 限速实现：\n",
    "  - 使用`pv -L 1m`限制接收速率为1MB/s\n",
    "  - 高低水位回调频繁触发（约每秒1次）\n",
    "- 控制效果：\n",
    "  - 实际吞吐稳定在954kB/s左右\n",
    "  - 停止发送后缓冲区数据持续排空"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 竞态条件及修复"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 竞态条件现象：单线程非阻塞TCP relay程序出现数据流停滞现象，表现为高水位回调后未触发写完成回调\n",
    "- 特殊性：通常认为竞态条件仅出现在多线程程序中，但单线程事件驱动模型同样可能出现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP Replay\n",
    "\n",
    "```shell\n",
    "bin/tcpreplay 127.0.0.1 3000 2000 # 监听2000, 数据发送到3000\n",
    "```\n",
    "\n",
    "- Server\n",
    "\n",
    "```shell\n",
    "nc -l -p 3000 | pv > /dev/null # 监听3000\n",
    "```\n",
    "\n",
    "- Client\n",
    "\n",
    "```shell\n",
    "nc localhost 2000 < /dev/zero\n",
    "```\n",
    "\n",
    "```shell\n",
    "utop\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 复现步骤：\n",
    "  - 启动存在bug的旧版本程序\n",
    "  - 建立client-server连接后数据流会异常停止\n",
    "  - CPU占用率降为0，程序进入停滞状态\n",
    "- 关键现象：高水位回调触发后未按预期调用写完成回调"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 使用GDB进行调试\n",
    "\n",
    "```shell\n",
    "gdb bin/tcpreplay 13409\n",
    "```\n",
    "\n",
    "- 调试发现：\n",
    "  - 程序阻塞在`epoll_wait`系统调用\n",
    "  - 单线程模型下难以直接定位问题连接\n",
    "  - 通过触发另一方向数据流可辅助诊断\n",
    "- 关键状态：TCP连接对象的`reading`标志为`FALSE`，表示已停止读取数据\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "继续触发断点\n",
    "\n",
    "- 调试技巧：\n",
    "  - 手动调用`startRead()`可临时恢复数据流, `p this->startRead()`\n",
    "  - `reading`标志会再次被设为`FALSE`导致停滞\n",
    "  - 需要通过日志和代码分析根本原因"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Race condition in non-blocking TCP relay\n",
    "\n",
    "- Two events happen in the same EventLoop iteration\n",
    "  - *C* is readable, call `S.send`,*S* is full, queue high watermark callback\n",
    "    - HWM and WC are deferred callbacks to prevent reentrant sendInLoop\n",
    "  - *S* is writable,*S* sends all data, but WC callback is not registered\n",
    "  - HWM is called, stop reading and register WC, WC will never be called\n",
    "- Tweak EventLoop to help debugging\n",
    "  - Single event model, `epoll_wait(maxevents=1)`\n",
    "  - Shuffle `activeChannels_` before calling `Channel::handleEvent()`\n",
    "  - Shuffle `pendingFunctors_ `in `EventLoop::doPendingFunctors()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 看不懂..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 原因分析"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 根本原因：\n",
    "  - 同一事件循环迭代中两个关联事件顺序异常：\n",
    "    - 连接*C*可读，调用`S.send()`导致队列满，排队高水位回调\n",
    "    - 连接*S*可写，发送完数据但写完成回调未注册\n",
    "  - 高水位回调停止读取并注册写完成回调，但后者永远不会被调用\n",
    "- 解决方案：\n",
    "  - 在高水位回调中检查输出缓冲区是否真的非空\n",
    "  - 实现事件处理的三种改进思路：\n",
    "    - 单事件模式, `epoll_wait(maxevents=1)`\n",
    "    - 处理前随机打乱活跃通道顺序\n",
    "    - 随机打乱待处理函数队列顺序\n",
    "- 调试建议：\n",
    "  - 可尝试将`pendingFunctors_`队列改为栈结构\n",
    "  - 通过改变事件处理顺序暴露潜在逻辑错误"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SOCKS4a服务器实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## SOCKS4 and SOCKS4a"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP relay has a fixed destination IP address\n",
    "  - TCP Balancer has a set ofdestination IP addresses\n",
    "- SOCKS4 let client specify the destination IP address\n",
    "- SOCKS4a let client specify the destination hostname\n",
    "  - SOCKS4a server does DNS resolution on behalf of client\n",
    "- Not ideal: blocking on DNS resolution\n",
    "  - Asynchronous DNS is possible with UDNS or c-ares (§7.15)\n",
    "  - To use it, split the `onServerMessage()` to two fragments: parse then AsyncDNSLookup, continue in DNSResolved callback"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## SOCKS4和SOCKS4a服务器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP relay特点：每收到一个客户连接，会向固定服务器IP地址发起请求\n",
    "- TCP Balancer特点：可以向一组服务器地址中的某个发起请求，实现负载均衡\n",
    "- SOCKS4特性：客户端可指定目标服务器的IP地址和端口\n",
    "- SOCKS4a改进：\n",
    "  - 客户端可指定主机名称(hostname)\n",
    "  - 服务器代替客户端进行DNS解析（解决客户端可能无法访问DNS的问题）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## SOCKS4a服务器实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "协议处理流程\n",
    "\n",
    "- 协议结构：\n",
    "  - 包含固定长度字段（版本号、命令、端口、IP等）\n",
    "  - 通过查找'\\0'字符定位`hostname`字段\n",
    "- DNS解析实现：\n",
    "  - 在69行代码处进行`hostname`解析\n",
    "  - 将解析得到的IP地址用于后续tunnel建立\n",
    "  - 解析失败时直接返回终止连接"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/socks4a/socks4a.cc\n",
    "\n",
    "#include \"examples/socks4a/tunnel.h\"\n",
    "\n",
    "#include \"muduo/net/Endian.h\"\n",
    "#include <stdio.h>\n",
    "#include <netdb.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "using namespace muduo;\n",
    "using namespace muduo::net;\n",
    "\n",
    "EventLoop* g_eventLoop;\n",
    "std::map<string, TunnelPtr> g_tunnels;\n",
    "\n",
    "void onServerConnection(const TcpConnectionPtr& conn)\n",
    "{\n",
    "  LOG_DEBUG << conn->name() << (conn->connected() ? \" UP\" : \" DOWN\");\n",
    "  if (conn->connected())\n",
    "  {\n",
    "    conn->setTcpNoDelay(true);\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    std::map<string, TunnelPtr>::iterator it = g_tunnels.find(conn->name());\n",
    "    if (it != g_tunnels.end())\n",
    "    {\n",
    "      it->second->disconnect();\n",
    "      g_tunnels.erase(it);\n",
    "    }\n",
    "  }\n",
    "}\n",
    "\n",
    "void onServerMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp)\n",
    "{\n",
    "  LOG_DEBUG << conn->name() << \" \" << buf->readableBytes();\n",
    "  if (g_tunnels.find(conn->name()) == g_tunnels.end())\n",
    "  {\n",
    "    if (buf->readableBytes() > 128)\n",
    "    {\n",
    "      conn->shutdown();\n",
    "    }\n",
    "    else if (buf->readableBytes() > 8)\n",
    "    {\n",
    "      const char* begin = buf->peek() + 8;\n",
    "      const char* end = buf->peek() + buf->readableBytes();\n",
    "      const char* where = std::find(begin, end, '\\0');\n",
    "      if (where != end)\n",
    "      {\n",
    "        char ver = buf->peek()[0];\n",
    "        char cmd = buf->peek()[1];\n",
    "        const void* port = buf->peek() + 2;\n",
    "        const void* ip = buf->peek() + 4;\n",
    "\n",
    "        sockaddr_in addr;\n",
    "        memZero(&addr, sizeof addr);\n",
    "        addr.sin_family = AF_INET;\n",
    "        addr.sin_port = *static_cast<const in_port_t*>(port);\n",
    "        addr.sin_addr.s_addr = *static_cast<const uint32_t*>(ip);\n",
    "\n",
    "        bool socks4a = sockets::networkToHost32(addr.sin_addr.s_addr) < 256;\n",
    "        bool okay = false;\n",
    "        if (socks4a)\n",
    "        {\n",
    "          const char* endOfHostName = std::find(where+1, end, '\\0');\n",
    "          if (endOfHostName != end)\n",
    "          {\n",
    "            string hostname = where+1;\n",
    "            where = endOfHostName; // 进行`hostname`解析\n",
    "            LOG_INFO << \"Socks4a host name \" << hostname;\n",
    "            InetAddress tmp; \n",
    "            if (InetAddress::resolve(hostname, &tmp))\n",
    "            {\n",
    "              addr.sin_addr.s_addr = tmp.ipv4NetEndian();\n",
    "              okay = true;\n",
    "            }\n",
    "          }\n",
    "          else\n",
    "          {\n",
    "            return;\n",
    "          }\n",
    "        }\n",
    "        else\n",
    "        {\n",
    "          okay = true;\n",
    "        }\n",
    "\n",
    "        InetAddress serverAddr(addr);\n",
    "        if (ver == 4 && cmd == 1 && okay)\n",
    "        {\n",
    "          TunnelPtr tunnel(new Tunnel(g_eventLoop, serverAddr, conn));\n",
    "          tunnel->setup();\n",
    "          tunnel->connect();\n",
    "          g_tunnels[conn->name()] = tunnel;\n",
    "          buf->retrieveUntil(where+1);\n",
    "          char response[] = \"\\000\\x5aUVWXYZ\";\n",
    "          memcpy(response+2, &addr.sin_port, 2);\n",
    "          memcpy(response+4, &addr.sin_addr.s_addr, 4);\n",
    "          conn->send(response, 8);\n",
    "        }\n",
    "        else\n",
    "        {\n",
    "          char response[] = \"\\000\\x5bUVWXYZ\";\n",
    "          conn->send(response, 8);\n",
    "          conn->shutdown();\n",
    "        }\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "  else if (!conn->getContext().empty())\n",
    "  {\n",
    "    const TcpConnectionPtr& clientConn\n",
    "      = boost::any_cast<const TcpConnectionPtr&>(conn->getContext());\n",
    "    clientConn->send(buf);\n",
    "  }\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  if (argc < 2)\n",
    "  {\n",
    "    fprintf(stderr, \"Usage: %s <listen_port>\\n\", argv[0]);\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    LOG_INFO << \"pid = \" << getpid() << \", tid = \" << CurrentThread::tid();\n",
    "\n",
    "    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));\n",
    "    InetAddress listenAddr(port);\n",
    "\n",
    "    EventLoop loop;\n",
    "    g_eventLoop = &loop;\n",
    "\n",
    "    TcpServer server(&loop, listenAddr, \"Socks4\");\n",
    "\n",
    "    server.setConnectionCallback(onServerConnection);\n",
    "    server.setMessageCallback(onServerMessage);\n",
    "\n",
    "    server.start();\n",
    "\n",
    "    loop.loop();\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "与TCP relay的区别\n",
    "\n",
    "- 主要差异：\n",
    "  - `onServerMessage`函数显著增大\n",
    "  - 增加了hostname解析功能\n",
    "  - 其他部分与TCP relay实现类似"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "阻塞式DNS的问题\n",
    "\n",
    "- 问题分析：\n",
    "  - 当前实现采用阻塞式DNS解析\n",
    "  - 解析可能耗时0.5秒至数秒（特别是递归查询时）\n",
    "  - 在单线程程序中会导致线程失去响应\n",
    "- 改进方案：\n",
    "  - 使用异步DNS库（UDNS或c-ares）\n",
    "  - 将`onServerMessage`拆分为两部分：\n",
    "    - 解析请求并提取hostname\n",
    "    - 在DNS回调中继续建立tunnel\n",
    "  - 需要保存和恢复执行上下文\n",
    "- 实现难点：\n",
    "  - 基于事件的回调式编程复杂度高\n",
    "  - 代码可读性会降低"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 非阻塞IO之外的选择"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Non-blocking IO is not the only choice"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Less \"context switches\" with blocking IO\n",
    "  - Code is much easier to follow and maintain, try reading Go's TLS code\n",
    "  - Naturally works better for bandwidth mismatch"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 实现差异：非阻塞IO需要开发者手动管理状态机，而阻塞IO由操作系统自动处理上下文切换\n",
    "- 性能对比：操作系统级上下文切换经过专家优化，通常比用户态实现的效率更高\n",
    "- 代码复杂度：异步DNS解析等操作会显著增加非阻塞IO代码的复杂度，如socket服务器实现案例所示"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 例题:Go语言TLS协议握手流程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- https://en.wikipedia.org/wiki/Transport_Layer_Security#Basic_TLS_handshake\n",
    "- https://pkg.go.dev/crypto/tls\n",
    "\n",
    "- 协议阶段：\n",
    "  - 协商阶段：ClientHello包含协议版本、随机数、密码套件等\n",
    "  - 证书交换：服务器发送Certificate消息（视密码套件可能省略）\n",
    "  - 密钥交换：`ServerKeyExchange`消息（DHE/DH_anon套件必需）\n",
    "- 实现对比：\n",
    "  - 非阻塞实现：需要维护复杂的状态机（等待`ClientHello`→等待`ServerHello`→等待`Certificate`等）\n",
    "  - Go实现：线性流程（`readHandshake`→处理`certMsg`→`write`响应等），如[`handshake_client.go`](https://cs.opensource.google/go/go/+/go1.25.1:src/crypto/tls/handshake_client.go)所示\n",
    "\n",
    "- 代码可读性：\n",
    "  - 非阻塞代码：需要分解为多个回调函数（如状态：等待hello→处理证书→等待密钥交换）\n",
    "  - Go代码：直线型流程（如`doFullHandshake`函数连续处理各阶段），更符合协议文档描述"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 例题:Go语言TCP relay项目"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- How to write efficient concurrent network programs with blocking IO? Use Go\n",
    "- https://github.com/chenshuo/muduo-examples-in-go\n",
    "  - examples/socks4a/tcprelay/tcprelay.go"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 并发模型优势：\n",
    "  - 传统线程：受限于栈大小（32位系统约几百线程/进程）\n",
    "  - Goroutine：轻量级线程（可创建数千个），解决并发度瓶颈\n",
    "- 编程范式：阻塞式代码更符合人类线性思维习惯"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 例题:TLS握手过程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://hpbn.co/transport-layer-security-tls/\n",
    "\n",
    "![8.7](./images/8.7.svg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 握手阶段：\n",
    "  - TCP三次握手（SYN→SYN-ACK→ACK）\n",
    "  - TLS协商（ClientHello→ServerHello→Certificate等）\n",
    "  - 密钥交换（ClientKeyExchange→ChangeCipherSpec）\n",
    "  - 完成握手（Finished消息）\n",
    "- 实现复杂度：\n",
    "  - 非阻塞实现：需要处理比TCP状态机更复杂的TLS状态转换\n",
    "  - Go实现：如[`handshake_server.go`](https://cs.opensource.google/go/go/+/refs/tags/go1.25.1:src/crypto/tls/handshake_server.go)所示，通过顺序调用`readClientHello`→`doResumeHandshake`等函数简化流程\n",
    "- 维护成本：\n",
    "  - 回调方式：需要跟踪多个分散的状态处理函数\n",
    "  - Go方式：集中式处理（如`serverHandshake`函数），更易调试和修改"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 用 GO 语言实现 TCP relay"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "go"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo-examples-in-go/blob/master/examples/socks4a/tcprelay/tcprelay.go\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"io\"\n",
    "\t\"log\"\n",
    "\t\"net\"\n",
    ")\n",
    "\n",
    "func panicOnError(err error) {\n",
    "\tif err != nil {\n",
    "\t\tpanic(err)\n",
    "\t}\n",
    "}\n",
    "\n",
    "func relay(serverconn net.Conn) {\n",
    "\tdefer serverconn.Close()\n",
    "\tlog.Printf(\"relay: listen on %v, new client from %v\\n\",\n",
    "\t\tserverconn.LocalAddr().String(), serverconn.RemoteAddr().String())\n",
    "\n",
    "\tclientconn, err := net.Dial(\"tcp\", \"localhost:3000\")\n",
    "\tpanicOnError(err)\n",
    "\tdefer clientconn.Close()\n",
    "\tlog.Printf(\"relay: connected to %v, from %v\\n\",\n",
    "\t\tclientconn.RemoteAddr().String(), clientconn.LocalAddr().String())\n",
    "\n",
    "\tdone := make(chan bool)\n",
    "\tgo func() {\n",
    "\t\tio.Copy(clientconn, serverconn)\n",
    "\t\tlog.Printf(\"relay: done copying serverconn to clientconn\\n\")\n",
    "\t\ttcpconn := clientconn.(*net.TCPConn)\n",
    "\t\ttcpconn.CloseWrite()\n",
    "\t\tdone <- true\n",
    "\t}()\n",
    "\n",
    "\tio.Copy(serverconn, clientconn)\n",
    "\tlog.Printf(\"relay: done copying clientconn to serverconn\\n\")\n",
    "\ttcpconn := serverconn.(*net.TCPConn)\n",
    "\ttcpconn.CloseWrite()\n",
    "\t<-done\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tlistener, err := net.Listen(\"tcp\", \":2000\")\n",
    "\tpanicOnError(err)\n",
    "\tdefer listener.Close()\n",
    "\tfor {\n",
    "\t\tconn, err := listener.Accept()\n",
    "\t\tif err != nil {\n",
    "\t\t\tcontinue // https://go.dev/src/net/http/server.go#L3467 func (s *Server) Serve(l net.Listener) error\n",
    "\t\t}\n",
    "\t\tgo relay(conn)\n",
    "\t}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## main函数讲解"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 基本结构: 主函数是一个无限循环，监听`2000`端口，接受连接后启动新的goroutine处理\n",
    "- 错误处理:\n",
    "  - 使用`panicOnError`处理监听错误\n",
    "  - 遇到`accept`错误时简单`continue`，实际应参考标准库的指数退避重试机制\n",
    "- 并发模型: 每个连接在独立goroutine中处理，相当于新建线程\n",
    "- 资源管理: 使用`defer`确保`listener`关闭"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 书 7.7.1 `accept()`时file descriptor耗尽"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## relay函数讲解"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 连接管理:\n",
    "  - 使用`defer`确保双端连接关闭\n",
    "  - 建立到localhost:3000的客户端连接\n",
    "- 数据转发:\n",
    "  - 双向转发：client→server和server→client\n",
    "  - 使用channel同步两个转发goroutine\n",
    "- 关闭机制:\n",
    "  - 任一方向传输完成即触发连接关闭\n",
    "  - 正确处理半关闭状态\n",
    "- 实现优势:\n",
    "  - 相比非阻塞IO实现更简洁\n",
    "  - 自动处理并发和资源释放"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP连接半关"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "go"
    }
   },
   "outputs": [],
   "source": [
    "// https://go.dev/src/net/tcpsock.go\n",
    "\n",
    "// CloseWrite shuts down the writing side of the TCP connection.\n",
    "// Most callers should just use Close.\n",
    "func (c *TCPConn) CloseWrite() error {\n",
    "\tif !c.ok() {\n",
    "\t\treturn syscall.EINVAL\n",
    "\t}\n",
    "\tif err := c.fd.closeWrite(); err != nil {\n",
    "\t\treturn &OpError{Op: \"close\", Net: c.fd.net, Source: c.fd.laddr, Addr: c.fd.raddr, Err: err}\n",
    "\t}\n",
    "\treturn nil\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- https://groups.google.com/g/golang-dev/c/cq-Y0vDXdwg\n",
    "\n",
    "```\n",
    "On Wed, Sep 28, 2011 at 11:27 AM, Russ Cox <r...@golang.org> wrote:\n",
    "> TCP supports this but I assert that protocols\n",
    "> that require it are broken.  What protocol do\n",
    "> you need this for?\n",
    "\n",
    "For example, some research protocols such as BMP.\n",
    "<http://tools.ietf.org/html/draft-ietf-grow-bmp-05>\n",
    "\n",
    "<quote>\n",
    "The monitored router MAY take steps to prevent the monitoring\n",
    "station from sending data (e.g. by half-closing the TCP session\n",
    "or setting its window size to zero) or it MAY silently discard any\n",
    "data erroneously sent by the monitoring station.\n",
    "</quote>\n",
    "\n",
    "There is a space even though very small and optional.\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 标准实践:\n",
    "  - 大多数情况直接使用Close而非Shutdown\n",
    "  - Shutdown专门用于半关闭场景\n",
    "- 历史背景:\n",
    "  - Go在1.0前加入Shutdown方法\n",
    "  - 主要用于特定协议如BMP监控\n",
    "- 实现细节:\n",
    "  - 调用closeWrite关闭写端\n",
    "  - 需要显式类型断言判断连接有效性"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP中继程序运行"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP Replay\n",
    "\n",
    "```shell\n",
    "go run tcpreplay.go \n",
    "```\n",
    "\n",
    "- Server\n",
    "\n",
    "```shell\n",
    "nc -l -p 3000 # 监听3000\n",
    "```\n",
    "\n",
    "- Client\n",
    "\n",
    "```shell\n",
    "nc localhost 2000\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 运行验证:\n",
    "  - 监听2000端口转发到3000端口\n",
    "  - 使用netcat作为客户端测试\n",
    "- 连接状态:\n",
    "  - 建立四条TCP连接(两个netcat各一个，relay两个)\n",
    "  - 通过netstat查看连接状态变化\n",
    "- 半关测试:\n",
    "  - 客户端`CTRL+D`触发半关闭\n",
    "  - 观察`FIN_WAIT2`和`CLOSE_WAIT`状态"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## strace半关连接"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "strace -tt nc -l -p 3000\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 测试方法:\n",
    "  - 使用strace追踪系统调用\n",
    "  - 客户端发送数据后挂起\n",
    "- 关闭过程:\n",
    "  - 服务端`read`返回`0`触发关闭\n",
    "  - shutdown写端发送FIN包\n",
    "  - 客户端恢复后完成全关闭\n",
    "- 关键观察:\n",
    "  - poll监控多个文件描述符\n",
    "  - 正确处理EOF条件\n",
    "  - 系统调用序列展示TCP状态转换"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 事件驱动与多线程的取舍"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Threads vs. Events debating for 20 years"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Why Threads Are A Bad Idea (for most purposes)\n",
    "  - 1995 John Ousterhout (Tcl/Tk, RAMCloud)\n",
    "  - [https://blog.acolyer.org/2014/12/09/why-threads-are-a-bad-idea/](https://web.stanford.edu/~ouster/cgi-bin/papers/threads.pdf)\n",
    "- Why Events Are A Bad Idea (for high-concurrency servers)\n",
    "\t- 2003 Rob von Behren, Jeremy Condit and Eric Brewer (CAP)\n",
    "  - [https://blog.acolyer.org/2014/12/10/why-events-are-a-bad-idea/](https://web.stanford.edu/class/cs240e/papers/threads-hotos-2003.pdf)\n",
    "- My rule of thumb\n",
    "  - Use threads (a.k.a. blocking IO) if you can\n",
    "  - Use events (a.k.a.non-blocking IO) if you must need\n",
    "\n",
    "\n",
    "> https://web.stanford.edu/class/archive/cs/cs240/cs240.1236/old//sp2020/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "争论起源: 计算机科学界关于线程和事件模型的争论持续了20年，涉及多位图灵奖得主和领域权威"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "线程是个坏主意\n",
    "- 代表人物: John Ousterhout（Tcl/Tk语言作者，RAMCloud项目负责人）\n",
    "- 核心观点: 1995年发表《Why Threads Are A Bad Idea》批评线程模型\n",
    "- 学术背景: 斯坦福大学教授，横跨工业界和学术界的计算机科学家\n",
    "- 批评依据: 线程编程复杂度高，容易引发竞态条件和死锁问题"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "事件是个坏主意\n",
    "- 反驳观点: 2003年Rob von Behren等人发表《Why Events Are A Bad Idea》\n",
    "- 代表人物: Eric Brewer（CAP理论提出者）\n",
    "- 核心论点: 事件驱动模型不适合高并发服务器场景\n",
    "- 学术地位: 两位作者均为分布式系统领域权威学者"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "个人看法\n",
    "\n",
    "- 选择原则:\n",
    "  - 优先使用线程模型（能用线程就用线程）\n",
    "  - 必要时才使用事件模型（必须用事件时才用）\n",
    "- 历史背景:\n",
    "  - 用户态线程支持始于1993年（Windows NT和Solaris）\n",
    "  - 并发编程研究始于1960年代（Dijkstra提出信号量和互斥器）\n",
    "  - 条件变量由Tony Hoare提出（均为图灵奖得主）\n",
    "- 技术演进:\n",
    "  - 60年代：操作系统内核开始使用线程\n",
    "  - 80年代：事件模型随BSD 4.2的TCP/IP协议栈出现\n",
    "  - 1995年：POSIX线程标准化\n",
    "  - 2006年：Linux 2.6内核完善线程库支持\n",
    "- 现代建议:\n",
    "  - 多线程编程应成为开发者必备技能\n",
    "  - 除非性能需求特殊，否则优先选择线程模型\n",
    "  - 事件模型适用于特定场景（如高性能网络编程）"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "name": "C++17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
